nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-18 15:12:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

magenta2: Introduce EPG

Browse Source
This commit is contained in:
Nirvana
2026-06-23 13:05:02 +02:00
parent a0167bb501
commit f23ab54059
3 changed files with 849 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magenta2/channel_manager.py
+146 -215
View File
@@ -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",
lib/streaming_providers/providers/magenta2/epg_manager.py
+625
View File
@@ -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,
)
lib/streaming_providers/providers/magenta2/provider.py
+78 -37
View File
@@ -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 #
# ------------------------------------------------------------------ #