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

Prepare recordings

Browse Source
This commit is contained in:
Nirvana
2026-03-23 17:46:10 +01:00
parent 66d11c1ba3
commit 3c1f24bb7c
2 changed files with 173 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magenta2/provider.py
+141 -28
View File
@@ -30,6 +30,7 @@ from .constants import (
DRM_SYSTEM_WIDEVINE,
ERROR_CODES,
MAGENTA2_CLIENT_IDS,
MAGENTA2_FALLBACK_ACCOUNT_URI,
MAGENTA2_LOGO,
MAGENTA2_PLATFORMS,
MODE_LIVE,
@@ -37,6 +38,14 @@ from .constants import (
SUPPORTED_COUNTRIES,
)
from .discovery import DiscoveryService
from ..lib_theplatform import (
TheplatformChannel,
fetch_distribution_rights,
fetch_entitled_channels_feed,
extract_persona_jwt,
build_licence_url,
build_widevine_drm_config,
)
from .endpoint_manager import EndpointManager
from .models import Magenta2Channel, Magenta2PlaybackRestrictedException
from .token_flow_manager import PersonaResult
@@ -314,6 +323,10 @@ class Magenta2Provider(StreamingProvider):
# get_manifest() / get_drm() can inject smil_base_url automatically
# without requiring callers (e.g. DRMOperations) to know about it.
self._recording_url_cache: Dict[str, str] = {}
# station_id → mpd_url / release_pid; populated by get_channels() so
# that get_manifest() / get_drm() can bypass SMIL for live channels.
self._live_manifest_cache: Dict[str, str] = {}
self._live_pid_cache: Dict[str, str] = {}
logger.info("Magenta2 provider initialization completed successfully")
@staticmethod
@@ -794,40 +807,84 @@ class Magenta2Provider(StreamingProvider):
**kwargs,
) -> List[StreamingChannel]:
"""
Fetch available channels from Magenta2 API
Fetch available channels from Magenta2 via the entitled-channels flow.
Uses lib_theplatform to:
1. Call getApplicableDistributionRights (license_service_url from manifest).
2. Fetch the entitled-channels feed filtered by those rights.
3. Convert each TheplatformChannel to a StreamingChannel.
"""
try:
headers = self._get_api_headers(require_auth=False)
cid = f"{self.session_id}::{self._generate_call_id()}"
user_agent = self.platform_config["user_agent"]
# Use the discovered channel stations endpoint
url = None
if self.endpoint_manager:
url = self.endpoint_manager.get_endpoint("channel_stations")
if not url:
url = self.endpoint_manager.get_endpoint("channel_list")
if not url and self.endpoint_manager.has_endpoint("mpx_feed_entitledChannelsFeed"):
url = self.endpoint_manager.get_endpoint("mpx_feed_entitledChannelsFeed")
# Final fallback
if not url:
url = "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-channel-stations-main"
url += "?lang=short-de&sort=dt%24displayChannelNumber&range=1-1000"
logger.debug(f"Fetching channels from: {url}")
response = self.http_manager.get(
url, operation="api", headers=headers, timeout=DEFAULT_REQUEST_TIMEOUT
# ── Step 1: resolve distribution rights ──────────────────────────
rights_url = (
self.provider_config.manifest.mpx.license_service_url
if self.provider_config and self.provider_config.manifest
else None
)
response.raise_for_status()
channels_data = response.json()
if not rights_url:
raise RuntimeError(
"No license_service_url available – configuration discovery may have failed"
)
channels = self._process_channel_stations_response_optimized(
channels_data, prefer_highest_quality
distribution_rights = fetch_distribution_rights(
http_manager=self.http_manager,
rights_url=rights_url,
cid=cid,
user_agent=user_agent,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
# ── Step 2: fetch entitled-channels feed ─────────────────────────
feed_url = (
self.endpoint_manager.get_endpoint("mpx_feed_entitledChannelsFeed")
if self.endpoint_manager
else None
) or "https://feed.entertainment.tv.theplatform.eu/f/mdeprod/mdeprod-entitled-channels"
tp_channels: List[TheplatformChannel] = fetch_entitled_channels_feed(
http_manager=self.http_manager,
feed_url=feed_url,
distribution_rights=distribution_rights,
cid=cid,
user_agent=user_agent,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
# ── Step 3: convert to StreamingChannel ──────────────────────────
channels: List[StreamingChannel] = []
for tp_ch in tp_channels:
try:
magenta2_channel = Magenta2Channel(
name=tp_ch.station_id, # placeholder; no display name in this feed
channel_id=tp_ch.station_id,
logo_url=None,
mode=MODE_LIVE,
content_type=CONTENT_TYPE_LIVE,
country=self.country,
raw_data=tp_ch.extra,
)
streaming_channel = magenta2_channel.to_streaming_channel(
provider_name=self.provider_name
)
streaming_channel.channel_number = tp_ch.channel_number
streaming_channel.manifest = tp_ch.mpd_url
if tp_ch.hls_url:
streaming_channel.hls_url = tp_ch.hls_url
# Cache manifest URL and releasePid for SMIL-free playback
self._live_manifest_cache[tp_ch.station_id] = tp_ch.mpd_url
self._live_pid_cache[tp_ch.station_id] = tp_ch.release_pid
channels.append(streaming_channel)
except Exception as exc:
logger.warning(f"get_channels: skipping channel {tp_ch.station_id}: {exc}")
logger.info(
f"Successfully fetched {len(channels)} channels for country {self.country} "
f"(quality preference: {'highest' if prefer_highest_quality else 'lowest'})"
f"Successfully fetched {len(channels)} entitled channels "
f"for country {self.country}"
)
return channels
@@ -1275,7 +1332,16 @@ class Magenta2Provider(StreamingProvider):
def get_manifest(
self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs
) -> Optional[str]:
"""Get MPD manifest URL via SmilManager."""
"""Get MPD manifest URL.
For live channels whose manifest was already fetched by get_channels(),
return the cached MPD URL directly without a SMIL round-trip.
VOD and recordings fall through to SmilManager as before.
"""
if content_type == CONTENT_TYPE_LIVE and content_id in self._live_manifest_cache:
logger.debug(f"get_manifest: cache hit for live channel {content_id}")
return self._live_manifest_cache[content_id]
if not self._smil_manager:
raise RuntimeError("SmilManager not available")
# Inject smil_base_url from recording cache when not already supplied.
@@ -1330,7 +1396,54 @@ class Magenta2Provider(StreamingProvider):
def get_drm(
self, content_id: str, content_type: str = CONTENT_TYPE_LIVE, **kwargs
) -> List[DRMConfig]:
"""Get DRM configuration via SmilManager."""
"""Get DRM configuration.
For live channels whose releasePid was cached by get_channels(), build
the Widevine licence URL directly using lib_theplatform — no SMIL fetch.
VOD and recordings fall through to SmilManager as before.
"""
if content_type == CONTENT_TYPE_LIVE and content_id in self._live_pid_cache:
release_pid = self._live_pid_cache[content_id]
logger.debug(
f"get_drm: building licence directly for live channel {content_id} "
f"(releasePid: {release_pid})"
)
try:
persona_token = self._ensure_authenticated()
raw_jwt = extract_persona_jwt(persona_token)
if not raw_jwt:
logger.error("get_drm: failed to extract persona JWT")
return []
widevine_endpoint = (
self.endpoint_manager.get_endpoint("widevine_license")
if self.endpoint_manager
else None
)
if not widevine_endpoint:
logger.error("get_drm: no widevine_license endpoint available")
return []
account_uri = (
self.provider_config.manifest.mpx.get_account_uri()
if self.provider_config and self.provider_config.manifest
else None
) or MAGENTA2_FALLBACK_ACCOUNT_URI
licence_url = build_licence_url(
widevine_endpoint=widevine_endpoint,
release_pid=release_pid,
persona_jwt=raw_jwt,
account_uri=account_uri,
)
return [build_widevine_drm_config(
licence_url=licence_url,
user_agent=self.platform_config["user_agent"],
)]
except Exception as exc:
logger.error(f"get_drm: direct licence build failed for {content_id}: {exc}")
return []
if not self._smil_manager:
raise RuntimeError("SmilManager not available")
# Same smil_base_url injection as get_manifest — keeps both consistent.
lib/streaming_providers/providers/magentaeu/provider.py
+32 -72
View File
@@ -5,11 +5,17 @@ import datetime
from typing import ClassVar, Dict, List, Optional
from ...base.auth import UserPasswordCredentials
from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel, Event
from ...base.models import DRMConfig, StreamingChannel, Event
from ...base.models.proxy_models import ProxyConfig
from ...base.network import ProxyConfigManager
from ...base.provider import StreamingProvider
from ...base.utils.logger import logger
from ..lib_theplatform import (
build_catchup_url,
build_licence_url,
build_widevine_drm_config,
parse_bifrost_epg_channel,
)
from .auth import MagentaAuthenticator
from .constants import (
API_ENDPOINTS,
@@ -213,40 +219,38 @@ class MagentaEUProvider(StreamingProvider):
channels = []
for channel_data in response_data["channels"]:
try:
# Extract channel information
tp_channel = parse_bifrost_epg_channel(channel_data)
if not tp_channel:
continue
# Bifrost-specific fields not covered by TheplatformChannel
title = channel_data.get("title", "Unknown Channel")
logo = channel_data.get("channel_logo", "")
manifest = channel_data.get("video_src_dash", "")
pid = channel_data.get("pid_dash", "")
station_id = channel_data.get("station_id", "")
channel_number = channel_data.get("channel_number", "")
media_pid = channel_data.get("media_pid", "")
is_audio = channel_data.get("is_audio", False)
# Build manifest script
# Build manifest script from bifrost metadata fields
manifest_script_parts = []
if channel_number:
manifest_script_parts.append(f"chno={channel_number}")
if station_id:
manifest_script_parts.append(f"epgid={station_id}")
if tp_channel.channel_number:
manifest_script_parts.append(f"chno={tp_channel.channel_number}")
if tp_channel.station_id:
manifest_script_parts.append(f"epgid={tp_channel.station_id}")
if media_pid:
manifest_script_parts.append(f"media={media_pid}")
manifest_script = " ".join(manifest_script_parts) if manifest_script_parts else ""
# Create streaming channel
streaming_channel = StreamingChannel(
name=title,
content_id=station_id or pid or title,
content_id=tp_channel.station_id or tp_channel.release_pid or title,
provider=self.provider_name,
logo_url=logo,
mode="live",
session_manifest=False,
manifest=manifest,
manifest=tp_channel.mpd_url,
manifest_script=manifest_script,
cdm_type=DRM_SYSTEM_WIDEVINE,
use_cdm=True,
cdm=f"pid={pid}" if pid else "",
cdm=f"pid={tp_channel.release_pid}" if tp_channel.release_pid else "",
cdm_mode="external",
video="best",
on_demand=True,
@@ -257,7 +261,6 @@ class MagentaEUProvider(StreamingProvider):
language=get_language(self.country),
streaming_format=STREAMING_FORMAT_DASH,
)
channels.append(streaming_channel)
except Exception as e:
@@ -348,38 +351,22 @@ class MagentaEUProvider(StreamingProvider):
Returns:
Manifest URL with catchup time parameters, or None if channel not found
"""
from ...base.utils.timestamp_converter import TimestampConverter
# Ensure cache is populated first
if not self._ensure_channels_cache():
logger.warning(f"Cannot get catchup manifest for {channel_id}, channels cache unavailable")
return None
# First get the base manifest URL
base_manifest = self.get_manifest(channel_id, **kwargs)
if not base_manifest:
logger.warning(f"Channel {channel_id} not found or has no manifest")
return None
try:
# Convert epoch seconds to ISO basic format (YYYYMMDDTHHMMSS)
start_iso = TimestampConverter.epoch_to_iso(
start_time, format_type="basic", as_utc=True
)
end_iso = TimestampConverter.epoch_to_iso(end_time, format_type="basic", as_utc=True)
# Build the catchup manifest URL
# Check if the manifest already has query parameters
separator = "&" if "?" in base_manifest else "?"
catchup_manifest = f"{base_manifest}{separator}begin={start_iso}&end={end_iso}"
catchup_manifest = build_catchup_url(base_manifest, start_time, end_time)
logger.debug(f"Catchup manifest for channel {channel_id}: {catchup_manifest}")
return catchup_manifest
except Exception as e:
logger.error(f"Error building catchup manifest for channel {channel_id}: {e}")
# Fall back to live manifest if catchup formatting fails
logger.warning(f"Falling back to live manifest for channel {channel_id}")
return base_manifest
@@ -416,10 +403,6 @@ class MagentaEUProvider(StreamingProvider):
def get_drm_config(self, channel: StreamingChannel, **kwargs) -> Optional[DRMConfig]:
"""Get DRM configuration for channel with correct authentication"""
try:
import base64
import json
from urllib.parse import quote # For URL encoding
from .auth import MagentaAuthToken, decode_jwt
from .constants import ACC_URL
@@ -443,16 +426,12 @@ class MagentaEUProvider(StreamingProvider):
logger.warning("No bearer token available for DRM config")
return None
# Remove 'Bearer ' prefix if present
if access_token.startswith("Bearer "):
access_token = access_token[7:]
# Decode JWT token to get account details
# Decode JWT to extract account ID and persona token
try:
# Get current token from authenticator
current_token = self.authenticator.current_token
# Use the helper method if token is MagentaAuthToken
if isinstance(current_token, MagentaAuthToken) and hasattr(
current_token, "get_jwt_claims"
):
@@ -461,14 +440,11 @@ class MagentaEUProvider(StreamingProvider):
logger.warning("Failed to get JWT claims from token")
return None
else:
# Fallback: use decode_jwt helper
decoded_payload = decode_jwt(access_token, verify=False)
except Exception as e:
logger.warning(f"Error decoding JWT token for DRM: {e}")
return None
# Extract account information from JWT payload
account_id = decoded_payload.get("dc_cts_accountId", "")
persona_token = decoded_payload.get("dc_cts_personaToken", "")
@@ -476,36 +452,20 @@ class MagentaEUProvider(StreamingProvider):
logger.warning("Missing account ID or persona token in JWT payload")
return None
# CORRECT: Build account URI and URL-encode it
account_uri = f"{ACC_URL}/{account_id}"
encoded_account_uri = quote(account_uri, safe="") # URL encode, NOT base64
# Build license URL with parameters
license_url = (
f"{WV_URL}{pid}&" f"token={persona_token}&" f"account={encoded_account_uri}"
license_url = build_licence_url(
widevine_endpoint=WV_URL,
release_pid=pid,
persona_jwt=persona_token,
account_uri=account_uri,
)
logger.debug(f"License URL created: {license_url[:100]}...")
# Create DRM config with minimal headers
drm_config = DRMConfig(
system=DRMSystem.WIDEVINE,
priority=1,
license=LicenseConfig.create_with_req_data(
req_data_template="{CHA-RAW}",
server_url=license_url,
req_headers=json.dumps(
{
"User-Agent": USER_AGENT,
"Content-Type": "application/octet-stream",
"Origin": get_base_url(self.country),
"Referer": f"{get_base_url(self.country)}/",
}
),
use_http_get_request=False,
),
drm_config = build_widevine_drm_config(
licence_url=license_url,
user_agent=USER_AGENT,
origin=get_base_url(self.country),
)
logger.debug(f"DRM config created successfully for channel {channel.name}")
return drm_config
@@ -579,4 +539,4 @@ class MagentaEUProvider(StreamingProvider):
elif country:
return f"Magenta TV ({country.upper()})"
else:
return "Magenta TV"
return "Magenta TV"