diff --git a/lib/streaming_providers/providers/magenta2/channel_manager.py b/lib/streaming_providers/providers/magenta2/channel_manager.py index 983fcef..180b857 100644 --- a/lib/streaming_providers/providers/magenta2/channel_manager.py +++ b/lib/streaming_providers/providers/magenta2/channel_manager.py @@ -3,20 +3,9 @@ """ Manages channel discovery, metadata enrichment, entitlement, and streaming-data population for the Magenta2 provider. - -Responsibilities ----------------- -- Fetch and cache the unauthenticated channel-stations feed (_fetch_station_metadata) -- Build StreamingChannel objects from theplatform feed entries -- Fetch the entitled-channels feed and merge it with station metadata (get_channels) -- Request entitlement tokens and playlist data per channel -- Populate streaming data (manifest URL, DRM) for a set of channels - -The live-manifest / live-pid caches live here because they are produced by -get_channels() and consumed by PlaybackManager via the provider's cache -attributes (_live_manifest_cache, _live_pid_cache). """ import json +import re import time from typing import Callable, Dict, List, Optional @@ -47,32 +36,15 @@ class ChannelManager: """ Handles all channel-related operations for Magenta2. - Parameters - ---------- - http_manager: - Shared HTTP manager (created by the provider). - provider_name: - Provider name string used when building StreamingChannel objects. - country: - Two-letter country code. - platform_config: - Platform-specific dict from MAGENTA2_PLATFORMS (user_agent, etc.). - session_id: - Session UUID shared with the provider instance. - serial_number: - Device serial UUID shared with the provider instance. - endpoint_manager: - Populated EndpointManager after discovery. - provider_config: - ProviderConfig after discovery. - auth_callback: - Callable[[], str] — returns a valid persona token (Basic-auth value). - Provided by the provider as ``self._ensure_authenticated``. - build_scaled_image_url_callback: - Callable[[str], Optional[str]] — scales a logo URL. - Provided by the provider. + ID Model (matches MagentaEU pattern): + - content_id = station_id (numeric ID from stationId URI) + - playback_id = opaque PID from era$mediaPids["urn:theplatform:tv:location:any"] + - Both IDs are stored in manifest_script for later reference """ + # Pattern to extract numeric station ID from stationId URI + _STATION_ID_PATTERN = re.compile(r"/Station/(\d+)$") + def __init__( self, http_manager, @@ -99,15 +71,16 @@ class ChannelManager: self._build_scaled_image_url = build_scaled_image_url_callback self.catchup_window = catchup_window - # Populated on first get_channels() call; returned directly on subsequent calls. + # Populated on first get_channels() call self._cached_channels: Optional[List[StreamingChannel]] = None - # release_pid → mpd_url, keyed by release_pid (= channel_id after get_channels). - self._live_manifest_cache: Dict[str, str] = {} - # release_pid → release_pid (used as a fast membership test by PlaybackManager). - self._live_pid_cache: Dict[str, str] = {} + # station_id (numeric) -> playback_id (opaque PID) + self._station_to_playback_id: Dict[str, str] = {} - # Populated eagerly at provider init (unauthenticated call). + # release_pid -> mpd_url (for fast manifest lookup) + self._live_manifest_cache: Dict[str, str] = {} + + # Pre-fetched station metadata (unauthenticated) self.station_metadata: Dict[str, Dict] = self._fetch_station_metadata() # ------------------------------------------------------------------ # @@ -122,20 +95,8 @@ class ChannelManager: prefer_highest_quality: bool = True, **kwargs, ) -> List[StreamingChannel]: - """ - Fetch available channels via the entitled-channels flow. - - Uses lib_theplatform to: - 1. Call getApplicableDistributionRights (license_service_url from manifest). - 2. Fetch the entitled-channels feed filtered by those rights. - 3. Merge with station metadata pre-fetched at init time (unauthenticated). - 4. Convert each TheplatformChannel to a StreamingChannel. - - Results are cached after the first successful call; subsequent calls - return from cache without hitting the network. - """ if self._cached_channels: - logger.debug("get_channels: returning from cache (no network calls)") + logger.debug("get_channels: returning from cache") return list(self._cached_channels) try: @@ -150,9 +111,7 @@ class ChannelManager: else None ) if not rights_url: - raise RuntimeError( - "No license_service_url available – configuration discovery may have failed" - ) + raise RuntimeError("No license_service_url available") persona_token = self._ensure_authenticated() auth_headers = { @@ -201,11 +160,12 @@ class ChannelManager: timeout=DEFAULT_REQUEST_TIMEOUT, ) - # ── Step 3: merge with station metadata (pre-fetched at init) ──── + # ── Step 3: merge with station metadata and build channels ────── channels: List[StreamingChannel] = [] for tp_ch in tp_channels: try: - meta = self.station_metadata.get(tp_ch.station_id, {}) + station_id = self._extract_station_id(tp_ch.station_id) + meta = self.station_metadata.get(station_id, {}) name = meta.get("title") or tp_ch.station_id logo_url = meta.get("logo_url") quality = meta.get("quality") @@ -214,12 +174,33 @@ class ChannelManager: if meta.get("channel_number") is not None else tp_ch.channel_number ) + playback_id = meta.get("playback_id") or tp_ch.release_pid + + if not station_id or not playback_id: + logger.warning( + f"Skipping channel {name}: missing station_id={station_id}, " + f"playback_id={playback_id}" + ) + continue catchup_hours = getattr(tp_ch, 'catchup_hours', None) or self.catchup_window + # Build manifest_script with both IDs (like MagentaEU) + manifest_script_parts = [] + if channel_number: + manifest_script_parts.append(f"chno={channel_number}") + if station_id: + manifest_script_parts.append(f"station={station_id}") + if playback_id: + manifest_script_parts.append(f"pid={playback_id}") + manifest_script = " ".join(manifest_script_parts) if manifest_script_parts else "" + + # Store mapping for DRM lookup + self._station_to_playback_id[station_id] = playback_id + magenta2_channel = Magenta2Channel( name=name, - channel_id=tp_ch.release_pid, + channel_id=station_id, # content_id = station_id (matches MagentaEU) logo_url=logo_url, mode=MODE_LIVE, content_type=CONTENT_TYPE_LIVE, @@ -232,13 +213,16 @@ class ChannelManager: streaming_channel.channel_number = channel_number streaming_channel.quality = quality streaming_channel.manifest = tp_ch.mpd_url + streaming_channel.manifest_script = manifest_script + streaming_channel.cdm_type = DRM_SYSTEM_WIDEVINE + streaming_channel.cdm = f"pid={playback_id}" + streaming_channel.cdm_mode = "external" + streaming_channel.catchup_hours = catchup_hours + if tp_ch.hls_url: streaming_channel.hls_url = tp_ch.hls_url - streaming_channel.catchup_hours = catchup_hours - - self._live_manifest_cache[tp_ch.release_pid] = tp_ch.mpd_url - self._live_pid_cache[tp_ch.release_pid] = tp_ch.release_pid + self._live_manifest_cache[playback_id] = tp_ch.mpd_url channels.append(streaming_channel) except Exception as exc: @@ -256,16 +240,50 @@ class ChannelManager: except Exception as e: raise Exception(f"Error fetching channels from Magenta2 API: {e}") + def get_playback_id_for_station(self, station_id: str) -> Optional[str]: + """Get the playback ID (opaque PID) for a station ID.""" + return self._station_to_playback_id.get(station_id) + + def get_station_id_for_channel(self, channel: StreamingChannel) -> Optional[str]: + """Get the station ID from a channel's manifest_script.""" + if not channel.manifest_script: + return None + # Parse "station=123456" from manifest_script + for part in channel.manifest_script.split(): + if part.startswith("station="): + return part.split("=", 1)[1] + return None + + def get_playback_id_from_channel(self, channel: StreamingChannel) -> Optional[str]: + """Get the playback ID from a channel's manifest_script or cdm field.""" + # Try cdm field first + if channel.cdm and channel.cdm.startswith("pid="): + return channel.cdm.split("=", 1)[1] + # Fall back to manifest_script + if channel.manifest_script: + for part in channel.manifest_script.split(): + if part.startswith("pid="): + return part.split("=", 1)[1] + return None + def get_entitlement_token( self, content_id: str, content_type: str = CONTENT_TYPE_LIVE ) -> str: """ - Request an entitlement token for *content_id* using the persona token - (Basic auth). + Request an entitlement token for *content_id*. + + Note: content_id here is the station_id (numeric). We need to map + it to playback_id for the entitlement request. """ self._ensure_authenticated() + + # Map station_id -> playback_id + playback_id = self._station_to_playback_id.get(content_id) + if not playback_id: + raise ValueError(f"No playback ID found for station {content_id}") + headers = self._get_api_headers(require_auth=True) - payload = {"content_id": content_id, "content_type": content_type} + payload = {"content_id": playback_id, "content_type": content_type} url = ( self._endpoint_manager.get_endpoint("entitlement") @@ -274,7 +292,7 @@ class ChannelManager: ) try: - logger.debug(f"Requesting entitlement token for: {content_id}") + logger.debug(f"Requesting entitlement token for: {content_id} (playback: {playback_id})") response = self._http.post( url, operation="auth", @@ -325,12 +343,17 @@ class ChannelManager: def get_channel_playlist(self, channel_id: str, entitlement_token: str) -> Dict: """Fetch playlist data (manifest URL, licence URL, format) for a channel.""" + # channel_id here is the station_id, but the API expects playback_id + playback_id = self._station_to_playback_id.get(channel_id) + if not playback_id: + raise ValueError(f"No playback ID found for station {channel_id}") + if self._endpoint_manager and self._endpoint_manager.has_endpoint("channel_playlist"): url = self._endpoint_manager.get_endpoint("channel_playlist").format( - channel_id=channel_id + channel_id=playback_id ) else: - url = f"https://api.magentatv.de/v1/channel/{channel_id}/playlist" + url = f"https://api.magentatv.de/v1/channel/{playback_id}/playlist" headers = { "Authorization": f"Bearer {entitlement_token}", @@ -355,13 +378,6 @@ class ChannelManager: channels: List[StreamingChannel], max_retries: int = DEFAULT_MAX_RETRIES, ) -> List[StreamingChannel]: - """ - Populate manifest URL, DRM config, and streaming format for each channel - by fetching an entitlement token and playlist. - - Channels that are restricted or repeatedly fail are silently dropped from - the returned list. - """ self._ensure_authenticated() successful_channels = [] @@ -372,6 +388,7 @@ class ChannelManager: while retries < max_retries and not success and not is_restricted: try: + # Use channel.channel_id (station_id) for entitlement logger.debug( f"Getting entitlement token for: {channel.name} (attempt {retries + 1})" ) @@ -398,11 +415,15 @@ class ChannelManager: if manifest_url: channel.manifest = manifest_url - channel.cdm_type = DRM_SYSTEM_WIDEVINE - channel.cdm = f"pid={channel.channel_id}" channel.license_url = license_url channel.certificate_url = certificate_url channel.streaming_format = streaming_format + + # Update manifest cache + playback_id = self.get_playback_id_from_channel(channel) + if playback_id: + self._live_manifest_cache[playback_id] = manifest_url + logger.info(f"Streaming data populated for: {channel.name}") successful_channels.append(channel) success = True @@ -430,36 +451,51 @@ class ChannelManager: return successful_channels def invalidate_cache(self) -> None: - """Clear the in-memory channel and live-manifest caches.""" self._cached_channels = None self._live_manifest_cache.clear() - self._live_pid_cache.clear() logger.debug("ChannelManager: caches cleared") # ------------------------------------------------------------------ # # Internal helpers # # ------------------------------------------------------------------ # + @staticmethod + def _extract_station_id(station_uri: str) -> Optional[str]: + """Extract numeric station ID from a stationId URI.""" + if not station_uri: + return None + match = ChannelManager._STATION_ID_PATTERN.search(station_uri) + return match.group(1) if match else None + + @staticmethod + def _extract_channel_id_from_entry(entry: Dict) -> Optional[str]: + """Extract the opaque playback PID from era$mediaPids.""" + try: + stations = entry.get("stations", {}) + if not stations: + return None + station_id = next(iter(stations.keys())) + station_info = stations[station_id] + era_media_pids = station_info.get("era$mediaPids", {}) + channel_id = era_media_pids.get("urn:theplatform:tv:location:any") + if channel_id: + logger.debug(f"Extracted playback ID from era$mediaPids: {channel_id}") + return channel_id + return None + except Exception as e: + logger.warning(f"Error extracting playback ID from entry: {e}") + return None + def _fetch_station_metadata(self) -> Dict[str, Dict]: """ - Fetch the unauthenticated channel-stations feed and return a lookup map - keyed by the theplatform Station URI. + Fetch the unauthenticated channel-stations feed and build metadata. - The URI is the key of the ``stations`` dict in each feed entry, e.g. - ``http://data.entertainment.tv.theplatform.eu/…/Station/265808936224``. - This matches ``listings[0].stationId`` in the entitled-channels feed, - which is what ``TheplatformChannel.station_id`` contains after parsing. - - Note: ``era$mediaPids["urn:theplatform:tv:location:any"]`` is a short - opaque PID used for other purposes — it is NOT the mapping key. - - Each value dict contains: - title – display name (" - Main" suffix stripped) - logo_url – scaled logo URL or None - quality – "HD", "SD", etc. - channel_number – display channel number or None + Returns a dict keyed by station_id (numeric), each value containing: + title, logo_url, quality, channel_number, playback_id """ metadata: Dict[str, Dict] = {} + self._station_to_playback_id.clear() + try: url = None if self._endpoint_manager: @@ -488,6 +524,14 @@ class ChannelManager: station_uri = next(iter(stations.keys())) station_info = stations[station_uri] + station_id = self._extract_station_id(station_uri) + if not station_id: + continue + + playback_id = self._extract_channel_id_from_entry(entry) + if playback_id: + self._station_to_playback_id[station_id] = playback_id + title = ( station_info.get("title") or entry.get("title", "") ).replace(" - Main", "") @@ -504,21 +548,23 @@ class ChannelManager: channel_number = entry.get("dt$displayChannelNumber") - existing = metadata.get(station_uri) + existing = metadata.get(station_id) if not existing or QUALITY_RANK.get(quality, 1) > QUALITY_RANK.get( existing["quality"], 1 ): - metadata[station_uri] = { + metadata[station_id] = { "title": title, "logo_url": logo_url, "quality": quality, "channel_number": channel_number, + "playback_id": playback_id, } except Exception as exc: logger.debug(f"_fetch_station_metadata: skipping entry: {exc}") logger.debug( - f"_fetch_station_metadata: built metadata for {len(metadata)} stations" + f"_fetch_station_metadata: built metadata for {len(metadata)} stations, " + f"{len(self._station_to_playback_id)} playback mappings" ) except Exception as exc: logger.warning( @@ -527,122 +573,7 @@ class ChannelManager: return metadata - @staticmethod - def _extract_channel_id_from_entry(entry: Dict) -> Optional[str]: - """Extract the correct channel ID from ``era$mediaPids``.""" - try: - stations = entry.get("stations", {}) - if not stations: - return None - station_id = next(iter(stations.keys())) - station_info = stations[station_id] - era_media_pids = station_info.get("era$mediaPids", {}) - channel_id = era_media_pids.get("urn:theplatform:tv:location:any") - if channel_id: - logger.debug(f"Extracted channel ID from era$mediaPids: {channel_id}") - return channel_id - fallback_id = entry.get("guid") - if fallback_id: - logger.warning(f"Using fallback channel ID from guid: {fallback_id}") - return fallback_id - logger.warning("No channel ID found in entry") - return None - except Exception as e: - logger.warning(f"Error extracting channel ID from entry: {e}") - return None - - def _create_channel_from_entry( - self, entry: Dict, station_info: Dict, display_number - ) -> Optional[StreamingChannel]: - """Build a StreamingChannel from a raw feed entry.""" - try: - title = station_info.get("title") or entry.get("title", "Unknown Channel") - title = title.replace(" - Main", "") - - channel_id = self._extract_channel_id_from_entry(entry) - if not channel_id: - return None - - logo_url = None - thumbnails = station_info.get("thumbnails", {}) - for logo_type in ["stationLogo", "stationLogoColored"]: - if logo_type in thumbnails: - original_url = thumbnails[logo_type].get("url") - if original_url: - logo_url = self._build_scaled_image_url(original_url) - break - - magenta2_channel = Magenta2Channel( - name=title, - channel_id=channel_id, - logo_url=logo_url, - mode=MODE_LIVE, - content_type=CONTENT_TYPE_LIVE, - country=self._country, - raw_data=entry, - ) - streaming_channel = magenta2_channel.to_streaming_channel( - provider_name=self._provider_name - ) - streaming_channel.channel_number = display_number - streaming_channel.quality = station_info.get("dt$quality", "SD") - return streaming_channel - - except Exception as e: - logger.warning(f"Error creating channel from entry: {e}") - return None - - def _process_channel_stations_response_optimized( - self, response_data: Dict, prefer_highest_quality: bool = True - ) -> List[StreamingChannel]: - """Single-pass deduplication and channel construction from a raw feed response.""" - if "entries" not in response_data: - return [] - - best_entries: Dict = {} - channels: List[StreamingChannel] = [] - - for entry in response_data["entries"]: - try: - stations = entry.get("stations", {}) - if not stations: - continue - station_info = next(iter(stations.values())) - display_number = entry.get("dt$displayChannelNumber") - - if display_number is None: - channel = self._create_channel_from_entry( - entry, station_info, display_number - ) - if channel: - channels.append(channel) - continue - - quality = station_info.get("dt$quality", "SD") - current_rank = QUALITY_RANK.get(quality, 1) - - existing = best_entries.get(display_number) - if not existing: - best_entries[display_number] = (entry, station_info, current_rank) - else: - _, _, existing_rank = existing - if (prefer_highest_quality and current_rank > existing_rank) or ( - not prefer_highest_quality and current_rank < existing_rank - ): - best_entries[display_number] = (entry, station_info, current_rank) - - except Exception: - continue - - for display_number, (entry, station_info, _) in best_entries.items(): - channel = self._create_channel_from_entry(entry, station_info, display_number) - if channel: - channels.append(channel) - - return channels - def _get_api_headers(self, require_auth: bool = False) -> Dict[str, str]: - """Build standard API request headers.""" headers = { "User-Agent": self._platform_config["user_agent"], "Accept": "application/json", diff --git a/lib/streaming_providers/providers/magenta2/epg_manager.py b/lib/streaming_providers/providers/magenta2/epg_manager.py new file mode 100644 index 0000000..3a8da04 --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/epg_manager.py @@ -0,0 +1,625 @@ +# streaming_providers/providers/magenta2/epg_manager.py +# -*- coding: utf-8 -*- +""" +EPG manager for the Magenta2 provider. + +Uses the ThePlatform API: +- Schedule: mdeprod-all-channel-schedules +- Details: mdeprod-all-programs + +Design: +- 3-hour block fetching for schedules +- No in-memory caching (stateless) +- Batch grid optimization (fetches once per day, distributes to all channels) +- No authentication required (guest access) — only device/session IDs +""" + +from __future__ import annotations + +import re +import time +import uuid +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Set, Tuple + +from ...base.models.epg_models import EPGEntry, EPGProgramDetails +from ...base.utils.logger import logger +from .constants import DEFAULT_REQUEST_TIMEOUT + + +class Magenta2EpgManager: + """ + Fetches and normalises EPG grid + programme-details data from the + Magenta2 ThePlatform API. + """ + + # HTTP statuses that should never be retried + _NO_RETRY_STATUSES: Set[int] = {400, 401, 403, 404} + + # Pattern to extract the numeric station ID from a stationId URI + # e.g. "http://data.entertainment.tv.theplatform.eu/entertainment/data/Station/265809960047" + # -> "265809960047" + _STATION_ID_PATTERN = re.compile(r"/Station/(\d+)$") + + # Pattern to parse credit role from credit ID + # e.g. "telekom.de-030d1565-director-gnp_1022271" -> "director" + _CREDIT_ROLE_PATTERN = re.compile(r"-[a-z]+-([a-z]+)-") + + # Map credit role strings to bucket names + _ROLE_MAP = { + "director": "directors", + "scriptwriter": "writers", + "writer": "writers", + "producer": "producers", + "cast": "cast", + "actor": "cast", + "presenter": "presenter", + "host": "presenter", + } + + def __init__( + self, + endpoint_manager: Any, + provider_config: Any, + http_manager: Any, + authenticator: Any, + fetch_details: bool = True, + default_past_days: int = 7, + default_future_days: int = 13, + ) -> None: + self._endpoint_manager = endpoint_manager + self._provider_config = provider_config + self._http = http_manager + self._auth = authenticator + self._fetch_details = fetch_details + self._default_past_days = default_past_days + self._default_future_days = default_future_days + + self._schedule_feed_url = self._resolve_feed_url("allChannelSchedulesFeed") + self._programs_feed_url = self._resolve_feed_url("allProgramsFeedUrl") + self._location_id = self._get_location_id() + + logger.info( + f"[Magenta2EpgManager] Initialised: " + f"schedule_feed={self._schedule_feed_url is not None}, " + f"programs_feed={self._programs_feed_url is not None}, " + f"location_id={self._location_id is not None}, " + f"fetch_details={fetch_details}" + ) + + # ------------------------------------------------------------------ + # URL resolution helpers + # ------------------------------------------------------------------ + + def _resolve_feed_url(self, feed_name: str) -> Optional[str]: + if not self._provider_config or not self._provider_config.manifest: + return None + + feed_template = self._provider_config.manifest.mpx.feeds.get(feed_name) + if not feed_template: + return None + + account_pid = self._provider_config.manifest.mpx.account_pid + if not account_pid: + return None + + return feed_template.replace("{MpxAccountPid}", account_pid) + + def _get_location_id(self) -> Optional[str]: + if not self._provider_config or not self._provider_config.manifest: + return None + return self._provider_config.manifest.mpx.location_id_uri + + # ------------------------------------------------------------------ + # HTTP helpers + # ------------------------------------------------------------------ + + def _current_ids(self) -> Tuple[str, str]: + token = getattr(self._auth, "current_token", None) + device_id = getattr(token, "device_id", "") or "" + session_id = getattr(token, "session_id", "") or "" + return device_id, session_id + + def _guest_headers(self) -> Dict[str, str]: + device_id, session_id = self._current_ids() + headers = { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + ), + "Accept": "application/json", + "Content-Type": "application/json", + "x-dt-session-id": session_id, + "x-dt-call-id": str(uuid.uuid4()), + } + if device_id: + headers["x-dt-device-id"] = device_id + return headers + + def _get_with_retry(self, url: str, operation: str) -> Optional[Dict[str, Any]]: + headers = self._guest_headers() + max_retries = 3 + + for attempt in range(max_retries): + headers["x-dt-call-id"] = str(uuid.uuid4()) + try: + response = self._http.get( + url, + operation=operation, + headers=headers, + timeout=DEFAULT_REQUEST_TIMEOUT, + ) + response.raise_for_status() + return response.json() + except Exception as exc: + status = getattr(getattr(exc, "response", None), "status_code", None) + + if status in self._NO_RETRY_STATUSES: + logger.warning( + f"[Magenta2EpgManager] {operation} failed with non-retryable " + f"status {status}, giving up: {exc}" + ) + return None + + if attempt < max_retries - 1: + wait_time = 2 ** attempt + logger.warning( + f"[Magenta2EpgManager] {operation} attempt {attempt + 1} " + f"failed, retrying in {wait_time}s: {exc}" + ) + time.sleep(wait_time) + else: + logger.error( + f"[Magenta2EpgManager] {operation} failed after " + f"{max_retries} attempts: {exc}" + ) + + return None + + # ------------------------------------------------------------------ + # Parsing helpers + # ------------------------------------------------------------------ + + @staticmethod + def _ensure_tz(dt: datetime) -> datetime: + if dt.tzinfo is None: + return dt.replace(tzinfo=timezone.utc) + return dt + + @staticmethod + def _parse_timestamp(ts: Optional[Any]) -> Optional[int]: + if ts is None: + return None + try: + return int(ts) // 1000 + except (ValueError, TypeError): + return None + + @staticmethod + def _parse_episode_number(value: Any) -> Optional[int]: + if value is None: + return None + try: + n = int(value) + return n if n > 0 else None + except (ValueError, TypeError): + return None + + @staticmethod + def _extract_station_id(station_uri: str) -> Optional[str]: + """Extract numeric station ID from a stationId URI.""" + if not station_uri: + return None + match = Magenta2EpgManager._STATION_ID_PATTERN.search(station_uri) + return match.group(1) if match else None + + def _resolve_entry_channel_id(self, entry: Dict[str, Any]) -> Optional[str]: + station_id_uri = entry.get("stationId") + if not station_id_uri: + return None + return self._extract_station_id(station_id_uri) + + # ------------------------------------------------------------------ + # Credit parsing from credit IDs + # ------------------------------------------------------------------ + + @classmethod + def _parse_credit_role(cls, credit_id: str) -> Optional[str]: + if not credit_id: + return None + match = cls._CREDIT_ROLE_PATTERN.search(credit_id) + return match.group(1) if match else None + + @classmethod + def _parse_credit_names_from_ids( + cls, credit_ids: List[str] + ) -> Dict[str, Optional[List[str]]]: + """ + Parse credit IDs into role buckets. + + Credit IDs contain the role in their name pattern, but not the + actual person's name. For now, we store the credit ID itself + as a placeholder. + """ + buckets: Dict[str, Set[str]] = { + "cast": set(), + "directors": set(), + "producers": set(), + "writers": set(), + "presenter": set(), + } + + for credit_id in credit_ids or []: + role = cls._parse_credit_role(credit_id) + if not role: + continue + + bucket = cls._ROLE_MAP.get(role) + if bucket: + buckets[bucket].add(credit_id) + + return { + key: sorted(values) if values else None + for key, values in buckets.items() + } + + # ------------------------------------------------------------------ + # Programme details fetching + # ------------------------------------------------------------------ + + def _fetch_program_details(self, program_guid: str) -> Dict[str, Any]: + if not program_guid or not self._programs_feed_url: + return {} + + # Don't restrict fields - get everything available + url = ( + f"{self._programs_feed_url}" + f"?byGuid={program_guid}" + f"&cid={uuid.uuid4()}" + ) + + data = self._get_with_retry(url, operation="epg_program_details") + if not data: + return {} + + entries = data.get("entries", []) + return entries[0] if entries else {} + + # ------------------------------------------------------------------ + # Schedule fetching + # ------------------------------------------------------------------ + + def _fetch_day_schedules( + self, + date: datetime, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + ) -> Dict[str, Any]: + if not self._schedule_feed_url or not self._location_id: + return {} + + merged: Dict[str, Any] = {} + + utc_date = self._ensure_tz(date).astimezone(timezone.utc) + formatted = utc_date.strftime("%Y-%m-%d") + + start_hour, end_hour = 0, 24 + + if start_time: + utc_start = self._ensure_tz(start_time).astimezone(timezone.utc) + if utc_start.date() == utc_date.date(): + start_hour = (utc_start.hour // 3) * 3 + + if end_time: + utc_end = self._ensure_tz(end_time).astimezone(timezone.utc) + if utc_end.date() == utc_date.date(): + if utc_end.hour % 3 == 0 and utc_end.minute == 0 and utc_end.second == 0: + end_hour = (utc_end.hour // 3) * 3 + else: + end_hour = ((utc_end.hour // 3) + 1) * 3 + + logger.debug( + f"[Magenta2EpgManager] Fetching {formatted} hours {start_hour}-{end_hour} (UTC)" + ) + + # Request stationId and programme fields we need + fields = ( + "stationId," + "listings.program.guid,listings.program.title," + "listings.program.description,listings.program.secondaryTitle," + "listings.program.tvSeasonNumber,listings.program.tvSeasonEpisodeNumber," + "listings.program.year,listings.program.tags," + "listings.startTime,listings.endTime" + ) + + for hour_offset in range(start_hour, end_hour, 3): + url = ( + f"{self._schedule_feed_url}" + f"?byListingTime={formatted}T{hour_offset:02d}:00:00.000Z" + f"~{formatted}T{hour_offset + 3:02d}:00:00.000Z" + f"&byLocationId={self._location_id}" + f"&fields={fields}" + f"&cid={uuid.uuid4()}" + ) + + data = self._get_with_retry(url, operation=f"epg_schedule_offset_{hour_offset}") + if data and data.get("entries"): + merged[url] = data + + return merged + + # ------------------------------------------------------------------ + # Listing -> EPGEntry + # ------------------------------------------------------------------ + + @staticmethod + def _extract_icon(thumbnails: Dict[str, Any]) -> Optional[str]: + for preferred in ("posterWideNoTitle", "mainWide", "HighResLandscape"): + entry = thumbnails.get(preferred) + if entry: + url = entry.get("url") + if url: + return url + return None + + @staticmethod + def _extract_genre(tags: List[Dict[str, Any]]) -> Optional[str]: + for tag in tags or []: + scheme = tag.get("scheme", "") + if scheme in ("genre-primary", "category"): + title = tag.get("title") + if title: + return title + return None + + def _parse_item_to_entry( + self, + item: Dict[str, Any], + channel_id: str, + start: int, + end: int, + program: Dict[str, Any], + ) -> Optional[EPGEntry]: + program_guid = program.get("guid", "") + title = program.get("title", "Unknown") + + broadcast_id = EPGEntry.encode_broadcast_id("magenta2", channel_id, start) + + details: Dict[str, Any] = {} + credit_map = { + "cast": None, + "directors": None, + "producers": None, + "writers": None, + "presenter": None, + } + + if self._fetch_details and program_guid: + details = self._fetch_program_details(program_guid) + + credit_ids = details.get("dt$creditIds", []) + if credit_ids: + credit_map = self._parse_credit_names_from_ids(credit_ids) + + title = details.get("title") or title + description = details.get("description") or program.get("description") + + year = details.get("year") or program.get("year") + try: + year = int(year) if year is not None else None + except (ValueError, TypeError): + year = None + + season_number = self._parse_episode_number( + details.get("tvSeasonNumber") or program.get("tvSeasonNumber") + ) + episode_number = self._parse_episode_number( + details.get("tvSeasonEpisodeNumber") or program.get("tvSeasonEpisodeNumber") + ) + + thumbnails = details.get("thumbnails", {}) or {} + icon = self._extract_icon(thumbnails) + + tags = details.get("tags", []) or program.get("tags", []) + genre_description = self._extract_genre(tags) + + return EPGEntry( + broadcast_id=broadcast_id, + title=title, + start=start, + end=end, + program_id=program_guid, + description=description, + plot_outline=None, + episode_name=None, + original_title=program.get("secondaryTitle") or details.get("secondaryTitle"), + year=year, + icon=icon, + cast=credit_map["cast"], + directors=credit_map["directors"], + writers=credit_map["writers"], + genre=None, + genre_sub_type=None, + genre_description=genre_description, + season_number=season_number, + episode_number=episode_number, + episode_part_number=None, + star_rating=None, + parental_rating=None, + parental_rating_code=None, + first_aired=None, + imdb_number=None, + series_link=None, + flags=None, + ) + + # ------------------------------------------------------------------ + # Window resolution + # ------------------------------------------------------------------ + + def _resolve_window( + self, + start_time: Optional[datetime], + end_time: Optional[datetime], + ) -> Tuple[datetime, datetime]: + def _to_utc(dt: datetime) -> datetime: + return self._ensure_tz(dt).astimezone(timezone.utc) + + now_utc = datetime.now(tz=timezone.utc) + + if start_time is None and end_time is None: + date_from = (now_utc - timedelta(days=self._default_past_days)).replace( + hour=0, minute=0, second=0, microsecond=0 + ) + date_to = (now_utc + timedelta(days=self._default_future_days)).replace( + hour=23, minute=59, second=59, microsecond=0 + ) + elif start_time is not None and end_time is None: + date_from = _to_utc(start_time) + date_to = date_from.replace(hour=23, minute=59, second=59, microsecond=0) + elif start_time is None and end_time is not None: + date_to = _to_utc(end_time) + date_from = date_to.replace(hour=0, minute=0, second=0, microsecond=0) + else: + date_from = _to_utc(start_time) + date_to = _to_utc(end_time) + + if date_to <= date_from: + logger.warning( + "[Magenta2EpgManager] end_time is not after start_time — " + "extending to end of start day" + ) + date_to = date_from.replace(hour=23, minute=59, second=59, microsecond=0) + + return date_from, date_to + + @staticmethod + def _dates_in_window(date_from: datetime, date_to: datetime) -> List[datetime]: + dates: List[datetime] = [] + current = date_from.astimezone(timezone.utc).replace( + hour=0, minute=0, second=0, microsecond=0 + ) + utc_to = date_to.astimezone(timezone.utc) + while current.date() <= utc_to.date(): + dates.append(current) + current += timedelta(days=1) + return dates + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def get_epg_grid( + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + channel_ids: Optional[List[str]] = None, + **_kwargs: Any, + ) -> Dict[str, List[EPGEntry]]: + date_from, date_to = self._resolve_window(start_time, end_time) + ts_from, ts_to = int(date_from.timestamp()), int(date_to.timestamp()) + + schedule_blocks: Dict[str, Any] = {} + for date in self._dates_in_window(date_from, date_to): + blocks = self._fetch_day_schedules(date, start_time, end_time) + if blocks: + schedule_blocks.update(blocks) + + wanted = set(channel_ids) if channel_ids else None + + if not schedule_blocks: + return {cid: [] for cid in wanted} if wanted else {} + + grid: Dict[str, List[EPGEntry]] = {} + + for data in schedule_blocks.values(): + for entry in data.get("entries", []): + channel_id = self._resolve_entry_channel_id(entry) + if channel_id is None: + continue + if wanted is not None and channel_id not in wanted: + continue + + bucket = grid.setdefault(channel_id, []) + for item in entry.get("listings", []): + start = self._parse_timestamp(item.get("startTime")) + end = self._parse_timestamp(item.get("endTime")) + if start is None or end is None or end <= start: + continue + if end <= ts_from or start >= ts_to: + continue + + program = item.get("program", {}) or {} + parsed = self._parse_item_to_entry(item, channel_id, start, end, program) + if parsed: + bucket.append(parsed) + + for entries in grid.values(): + entries.sort(key=lambda p: p.start) + + if wanted is not None: + for cid in wanted: + grid.setdefault(cid, []) + + logger.info( + f"[Magenta2EpgManager] Grid EPG: " + f"{sum(len(v) for v in grid.values())} programmes across " + f"{len(grid)} channels" + ) + return grid + + def get_channel_epg( + self, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **_kwargs: Any, + ) -> List[EPGEntry]: + """Get EPG for a single channel (delegates to grid).""" + result = self.get_epg_grid( + start_time=start_time, + end_time=end_time, + channel_ids=[channel_id], + **_kwargs, + ) + return result.get(channel_id, []) + + def get_program_details(self, program_id: str) -> Optional[EPGProgramDetails]: + if not program_id: + return None + + raw = self._fetch_program_details(program_id) + if not raw: + return None + + credit_ids = raw.get("dt$creditIds", []) + credit_map = { + "cast": None, + "directors": None, + "producers": None, + "writers": None, + "presenter": None, + } + if credit_ids: + credit_map = self._parse_credit_names_from_ids(credit_ids) + + year = raw.get("year") + try: + year = int(year) if year is not None else None + except (ValueError, TypeError): + year = None + + return EPGProgramDetails( + program_id=program_id, + description=raw.get("description"), + episode_name=raw.get("secondaryTitle"), + year=year, + icon=self._extract_icon(raw.get("thumbnails", {}) or {}), + cast=credit_map["cast"], + directors=credit_map["directors"], + writers=credit_map["writers"], + producers=credit_map["producers"], + presenter=credit_map["presenter"], + composers=None, + contributors=None, + ) \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index 41dac9b..297cece 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -6,19 +6,21 @@ Magenta2 streaming provider. This module contains only lifecycle, authentication, and the thin public API that delegates to the three domain managers: - ChannelManager – channel discovery, entitlement, streaming-data population - PlaybackManager – manifest / DRM routing (live fast-path + SMIL fallback) - VodManager – VOD catalogue browsing + ChannelManager – channel discovery, entitlement, streaming-data population + PlaybackManager – manifest / DRM routing (live fast-path + SMIL fallback) + VodManager – VOD catalogue browsing RecordingsManager – nPVR (list / delete / manifest) - SmilManager – SMIL-based manifest and DRM for VOD / recordings + SmilManager – SMIL-based manifest and DRM for VOD / recordings + Magenta2EpgManager – EPG grid + programme-details (ThePlatform API) """ import uuid -from datetime import datetime, timedelta +from datetime import datetime from typing import Any, ClassVar, Dict, List, Optional, Tuple, cast from urllib.parse import quote from ...base.models import DRMConfig, StreamingChannel, Event from ...base.models.auth import AuthState +from ...base.models.epg_models import EPGEntry, EPGProgramDetails from ...base.models.proxy_models import ProxyConfig from ...base.network import HTTPManagerFactory, ProxyConfigManager from ...base.provider import StreamingProvider @@ -28,6 +30,7 @@ from .smil_manager import SmilManager from .vod_manager import VodManager from .auth import Magenta2Authenticator, Magenta2Credentials, Magenta2UserCredentials from .channel_manager import ChannelManager +from .epg_manager import Magenta2EpgManager from .playback_manager import PlaybackManager from .config_models import BootstrapConfig, ProviderConfig from .constants import ( @@ -182,6 +185,7 @@ class Magenta2Provider(StreamingProvider): self._vod_manager: Optional[VodManager] = None self._recordings_manager: Optional[RecordingsManager] = None self._smil_manager: Optional[SmilManager] = None + self._epg_manager: Optional[Magenta2EpgManager] = None self._vod_manager = VodManager( http_manager=self.http_manager, @@ -241,6 +245,18 @@ class Magenta2Provider(StreamingProvider): ) logger.info("✓ PlaybackManager initialized") + # ── EPG Manager ──────────────────────────────────────────────────────── + self._epg_manager = Magenta2EpgManager( + endpoint_manager=self.endpoint_manager, + provider_config=self.endpoint_manager.config, + http_manager=self.http_manager, + authenticator=self.authenticator, + fetch_details=True, # Fetch full programme details (credits, images, etc.) + default_past_days=7, + default_future_days=13, + ) + logger.info("✓ EPG Manager initialized") + # ── Auth bridge ─────────────────────────────────────────────────────── self.device_token = None self._auth = AuthBridge( @@ -297,8 +313,9 @@ class Magenta2Provider(StreamingProvider): return False @property - def implements_epg(self) -> bool: - return False + def epg_window(self) -> Tuple[int, int]: + """Return EPG window as (past_days, future_days).""" + return 7, 13 # 7 days past, 13 days future @property def implements_recordings(self) -> bool: @@ -741,38 +758,62 @@ class Magenta2Provider(StreamingProvider): start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, **kwargs: Any, - ) -> List[Dict[str, Any]]: - """Fetch EPG data for a channel.""" - try: - if start_time is None: - start_time = datetime.now() - if end_time is None: - end_time = datetime.now() + timedelta(hours=DEFAULT_EPG_WINDOW_HOURS) - - headers = self._get_api_headers(require_auth=False) - url: str = ( - self.endpoint_manager.get_endpoint("epg") - or "https://api.magentatv.de/proxy/device/epg" - ) - - params = { - "channelId": channel_id, - "start": start_time.isoformat(), - "end": end_time.isoformat(), - } - response = self.http_manager.get( - url, - operation="api", - headers=headers, - params=params, - timeout=DEFAULT_REQUEST_TIMEOUT, - ) - response.raise_for_status() - return response.json() # type: ignore[no-any-return] - except Exception as e: - logger.error(f"Error getting EPG for channel {channel_id}: {e}") + ) -> List[EPGEntry]: + """Get EPG data for a specific channel (delegates to Magenta2EpgManager).""" + if not self._epg_manager: + logger.warning(f"{self.provider_name}: EPG manager not initialized") return [] + try: + self._ensure_authenticated() + except Exception as e: + logger.warning(f"{self.provider_name}: Auth failed for EPG: {e}") + + return self._epg_manager.get_channel_epg( + channel_id=channel_id, + start_time=start_time, + end_time=end_time, + **kwargs, + ) + + def get_epg_grid( + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + channel_ids: Optional[List[str]] = None, + **kwargs: Any, + ) -> Dict[str, List[EPGEntry]]: + """Get EPG data for all channels (or a subset) over a time window.""" + if not self._epg_manager: + logger.warning(f"{self.provider_name}: EPG manager not initialized") + return {} + + try: + self._ensure_authenticated() + except Exception as e: + logger.warning(f"{self.provider_name}: Auth failed for EPG grid: {e}") + + # If channel_ids is None, pull the full channel list and use each + # channel's station-ID-derived channel_id (the same ID space the + # EPG manager keys its grid by — see ChannelManager / EPGEntry wiring). + if channel_ids is None: + channels = self._channel_manager.get_channels(populate_streaming=False) + channel_ids = [ch.channel_id for ch in channels] + + return self._epg_manager.get_epg_grid( + start_time=start_time, + end_time=end_time, + channel_ids=channel_ids, + **kwargs, + ) + + def get_program_details(self, program_id: str, **kwargs: Any) -> Optional[EPGProgramDetails]: + """Get detailed metadata for a single programme.""" + if not self._epg_manager: + logger.warning(f"{self.provider_name}: EPG manager not initialized") + return None + return self._epg_manager.get_program_details(program_id) + # ------------------------------------------------------------------ # # Auth state / readiness introspection # # ------------------------------------------------------------------ #