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

Prepare RTLPlus for new streaming prov

Browse Source
This commit is contained in:
Nirvana
2026-04-28 18:54:26 +02:00
parent 6001628d72
commit 96fe3b6c15
4 changed files with 1038 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/rtlplus/channel_manager.py
+74 -344
View File
@@ -11,14 +11,10 @@ Handles all linear TV (live channel) functionality:
"""
from typing import List, Optional, Dict, Any
import json
import urllib.parse
import time
from datetime import date
from ...base.models import Channel, DRMConfig, DRMSystem, LicenseConfig, LicenseUnwrapperParams
from ...base.models import Channel, DRMConfig
from ...base.utils.logger import logger
from .constants import RTLPlusDefaults
class RTLPlusChannelManager:
@@ -28,12 +24,8 @@ class RTLPlusChannelManager:
Uses the authenticator for all token management.
"""
# Cache TTL for channel layouts (5 minutes)
_CHANNEL_LAYOUT_CACHE_TTL = 300
def __init__(self, provider):
self._provider = provider
self._layout_cache: Dict[str, tuple] = {} # seo -> (data, timestamp)
@property
def cfg(self):
@@ -111,7 +103,6 @@ class RTLPlusChannelManager:
slug = value_layout.get("id") # e.g., "rtlde_rtl"
seo = value_layout.get("seo") # e.g., "rtl"
# Debug log the extraction
logger.debug(f"Extracted: title='{title}', slug='{slug}', seo='{seo}', logo_id='{logo_id}'")
if slug is None:
@@ -166,7 +157,7 @@ class RTLPlusChannelManager:
# Create channel with proper content_id
streaming_channel = Channel.create_live_channel(
name=title,
channel_id=content_id, # This sets content_id
channel_id=content_id,
provider=self._provider.provider_name,
)
@@ -174,12 +165,11 @@ class RTLPlusChannelManager:
streaming_channel.logo_url = self._resolve_image_url(logo_id)
# Store SEO in manifest_script for fallback lookups
# This allows get_manifest to work with either slug or seo
if seo:
streaming_channel.manifest_script = seo
logger.debug(
f"Created channel: name={title}, id={streaming_channel.content_id}, manifest_script={streaming_channel.manifest_script}")
logger.debug(f"Created channel: name={title}, id={streaming_channel.content_id}, "
f"manifest_script={streaming_channel.manifest_script}")
streaming_channels.append(streaming_channel)
@@ -193,119 +183,6 @@ class RTLPlusChannelManager:
return ""
return f"https://images.rtl.de/{image_id}?format=webp&width=200"
# --------------------------------------------------------------------------
# Channel Layout & Stream Assets
# --------------------------------------------------------------------------
def fetch_channel_layout(self, channel_seo: str, force_refresh: bool = False) -> Dict[str, Any]:
"""
Fetch the complete layout JSON for a linear TV channel with caching.
"""
now = time.time()
if not force_refresh:
cached = self._layout_cache.get(channel_seo)
if cached and (now - cached[1]) < self._CHANNEL_LAYOUT_CACHE_TTL:
logger.debug(f"Using cached layout for {channel_seo}")
return cached[0]
oauth_token = self._provider.get_user_bearer_token()
if not oauth_token:
raise RuntimeError("User authentication required to fetch channel layout")
# Get Bedrock token - no parameters needed
bedrock_token = self.auth.get_bedrock_token()
url = self.cfg.get_bedrock_layout_url(channel_seo=channel_seo)
location = f"{self.cfg.beta_website}{channel_seo}/live"
headers = self.cfg.get_bedrock_layout_headers(oauth_token, bedrock_token, location)
params = {"blockPage": 1, "nbPages": 2}
response = self.http.get(url, headers=headers, params=params, operation="api")
response.raise_for_status()
data = response.json()
self._layout_cache[channel_seo] = (data, now)
return data
def extract_stream_assets(self, channel_seo: str) -> List[Dict[str, Any]]:
"""Extract all stream assets from a channel's layout."""
layout = self.fetch_channel_layout(channel_seo)
# Log full response for debugging when no assets found
blocks = layout.get("blocks", [])
found_assets = False
for block in blocks:
if block.get("type") == "bffPaginated":
items = block.get("content", {}).get("items", [])
for item in items:
if item.get("itemType") == "video":
video = item.get("itemContent", {}).get("video", {})
assets = video.get("assets", [])
if assets:
return assets
elif item.get("itemType") == "classic":
# Some live layouts use 'classic' with a nested player block
item_content = item.get("itemContent", {})
video = item_content.get("video", {})
assets = video.get("assets", [])
if assets:
logger.debug(f"Found {len(assets)} assets in classic item for {channel_seo}")
return assets
if not found_assets:
# Log the entire response structure for debugging
logger.error(f"No stream assets found for channel {channel_seo}")
logger.error(f"Full response structure for {channel_seo}:")
# Log the high-level structure
logger.error(f"Response keys: {list(layout.keys())}")
# Log blocks information
logger.error(f"Number of blocks: {len(blocks)}")
for idx, block in enumerate(blocks):
block_type = block.get("type")
logger.error(f" Block {idx}: type={block_type}")
if block_type == "bffPaginated":
content = block.get("content", {})
items = content.get("items", [])
logger.error(f" items count: {len(items)}")
for item_idx, item in enumerate(items):
item_type = item.get("itemType")
logger.error(f" Item {item_idx}: itemType={item_type}")
if item_type == "video":
item_content = item.get("itemContent", {})
logger.error(f" itemContent keys: {list(item_content.keys())}")
video = item_content.get("video", {})
logger.error(f" video keys: {list(video.keys())}")
assets = video.get("assets", [])
logger.error(f" assets count: {len(assets)}")
# Log the actual video structure
logger.error(f" Full video object: {json.dumps(video, indent=2)[:1000]}")
# Also log the entity info if available
entity = layout.get("entity", {})
if entity:
logger.error(f"Entity info: id={entity.get('id')}, type={entity.get('type')}")
metadata = entity.get("metadata", {})
logger.error(f"Entity metadata: title={metadata.get('title')}, code={metadata.get('code')}")
logger.warning(f"No stream assets found for channel {channel_seo}")
return []
def invalidate_layout_cache(self, channel_seo: str = None):
"""Invalidate layout cache for a specific channel or all channels."""
if channel_seo:
self._layout_cache.pop(channel_seo, None)
else:
self._layout_cache.clear()
# --------------------------------------------------------------------------
# Channel ID Normalization
# --------------------------------------------------------------------------
@@ -327,7 +204,7 @@ class RTLPlusChannelManager:
return channel_id
# --------------------------------------------------------------------------
# Manifest & DRM
# Manifest & DRM (using provider's common methods)
# --------------------------------------------------------------------------
def get_best_manifest_url(
@@ -349,200 +226,44 @@ class RTLPlusChannelManager:
token = self._provider.get_user_bearer_token()
if not token:
logger.error("User authentication required for linear TV stream, not authenticated")
return None # or [] for DRM
return None
channel_seo = self._normalize_channel_identifier(channel_id)
assets = self.extract_stream_assets(channel_seo)
# Use provider's common layout fetching
layout = self._provider.fetch_layout(
layout_type="live",
content_id=channel_seo,
location=f"{self.cfg.beta_website}{channel_seo}/live"
)
quality_pref = preferred_quality or next(iter(self.cfg.preferred_qualities), "hd")
format_pref = preferred_format or next(iter(self.cfg.preferred_formats), "dashcenc")
drm_pref = preferred_drm_type or next(iter(self.cfg.preferred_drm_types), "hardware")
if not layout:
return None
# Try exact match
for asset in assets:
if (asset.get("quality") == quality_pref and
asset.get("format") == format_pref and
asset.get("drm", {}).get("type") == drm_pref):
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
logger.info(f"Found exact match manifest for {channel_seo}")
return manifest_url
# Extract assets using common method
assets = self._provider.extract_video_assets(layout)
# Try format + quality, any DRM
for asset in assets:
if asset.get("quality") == quality_pref and asset.get("format") == format_pref:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
logger.info(f"Found format/quality match manifest for {channel_seo}")
return manifest_url
# Extract best manifest URL using common method
return self._provider.extract_best_manifest_url(assets, preferred_quality, preferred_format)
# Fallback to any asset from preferred formats
for fmt in self.cfg.preferred_formats:
for qual in self.cfg.preferred_qualities:
for asset in assets:
if asset.get("format") == fmt and asset.get("quality") == qual:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
logger.info(f"Found fallback manifest for {channel_seo}")
return manifest_url
# Last resort: any manifest URL
for asset in assets:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
logger.warning(f"Using last-resort manifest for {channel_seo}")
return manifest_url
logger.error(f"No manifest URL found for channel {channel_seo}")
return None
def get_drm_config_for_channel(
self,
channel_id: str,
preferred_quality: str = None,
) -> List[DRMConfig]:
def get_drm_config_for_channel(self, channel_id: str) -> List[DRMConfig]:
"""
Get DRM configuration for a linear TV channel.
Supports both Widevine and PlayReady.
Uses delta provider with dashcenc format for the best compatibility.
Returns both DRM types using the same upfront token.
"""
channel_seo = self._normalize_channel_identifier(channel_id)
assets = self.extract_stream_assets(channel_seo)
# Priority: delta provider, then dashcenc format, then HD quality
quality = preferred_quality or "hd" # HD is primary from constants
# Find the best asset: delta provider + dashcenc + preferred quality
target_asset = None
# First try: delta + dashcenc + preferred quality
for asset in assets:
if (asset.get("provider") == "delta" and
asset.get("format") == "dashcenc" and
asset.get("quality") == quality):
target_asset = asset
logger.debug(f"Found delta/dashcenc/{quality} asset for {channel_seo}")
break
# Fallback: delta + dashcenc + any quality
if not target_asset:
for asset in assets:
if asset.get("provider") == "delta" and asset.get("format") == "dashcenc":
target_asset = asset
actual_quality = asset.get("quality", "unknown")
logger.debug(f"Fallback to delta/dashcenc/{actual_quality} for {channel_seo}")
break
# Last resort: any dashcenc asset
if not target_asset:
for asset in assets:
if asset.get("format") == "dashcenc":
target_asset = asset
provider = asset.get("provider", "unknown")
logger.debug(f"Last resort: {provider}/dashcenc for {channel_seo}")
break
if not target_asset:
logger.error(f"No dashcenc asset found for channel {channel_seo}")
return []
# Extract contentId from the selected asset
drm_info = target_asset.get("drm", {})
drm_config = drm_info.get("config", {})
content_id = drm_config.get("contentId")
if not content_id:
logger.error(f"No contentId in asset for {channel_seo}")
return []
try:
uid = self.auth.get_user_id_from_token()
if not uid:
logger.error(f"No user ID available for DRM on {channel_seo}")
return []
# Get ONE upfront token (works for both Widevine and PlayReady)
upfront_token = self.auth.get_upfront_token(
content_id=content_id,
uid=uid,
)
if not upfront_token:
logger.error(f"Failed to get upfront token for {channel_seo}")
return []
drm_configs = []
# Build Widevine config
wv_config = self._get_widevine_config(upfront_token)
if wv_config:
drm_configs.append(wv_config)
logger.debug(f"Added Widevine DRM for {channel_seo}")
# Build PlayReady config (same token works)
pr_config = self._get_playready_config(upfront_token)
if pr_config:
drm_configs.append(pr_config)
logger.debug(f"Added PlayReady DRM for {channel_seo}")
logger.info(f"Built {len(drm_configs)} DRM configs for {channel_seo} (Widevine + PlayReady)")
return drm_configs
except Exception as e:
logger.error(f"Failed to get DRM for {channel_seo}: {e}")
return []
def _get_widevine_config(self, upfront_token: str) -> Optional[DRMConfig]:
"""Get Widevine DRM configuration using existing headers from constants."""
if not upfront_token:
return None
# Get headers from config (which now includes origin/referer)
headers = self.cfg.get_drm_license_headers(upfront_token)
# URL-encode headers for req_headers parameter
req_headers = urllib.parse.urlencode(headers)
# Create LicenseConfig with unwrapper settings
license_config = LicenseConfig(
server_url=self.cfg.drmtoday_license_url,
req_headers=req_headers,
req_data="{CHA-RAW}",
use_http_get_request=False,
unwrapper="json,base64",
unwrapper_params=LicenseUnwrapperParams(path_data="license"),
# Use provider's common layout fetching
layout = self._provider.fetch_layout(
layout_type="live",
content_id=channel_seo,
location=f"{self.cfg.beta_website}{channel_seo}/live"
)
return DRMConfig(
system=DRMSystem.WIDEVINE,
priority=3,
license=license_config,
)
if not layout:
return []
def _get_playready_config(self, upfront_token: str) -> Optional[DRMConfig]:
"""Get PlayReady DRM configuration using existing headers from constants."""
if not upfront_token:
return None
# Use the existing method from RTLPlusConfig
headers = self.cfg.get_playready_license_headers(upfront_token)
# URL-encode headers for req_headers parameter
req_headers = urllib.parse.urlencode(headers)
return DRMConfig(
system=DRMSystem.PLAYREADY,
priority=2, # Slightly lower priority than Widevine (fallback)
license=LicenseConfig(
server_url=RTLPlusDefaults.DRMTODAY_PLAYREADY_URL,
req_headers=req_headers,
req_data="{CHA-RAW}", # PlayReady expects raw challenge
use_http_get_request=False,
),
)
# Delegate to provider's common DRM method
return self._provider.get_drm_for_content(layout)
# --------------------------------------------------------------------------
# Channel Info Helpers
@@ -551,48 +272,57 @@ class RTLPlusChannelManager:
def get_channel_info(self, channel_id: str) -> Optional[Dict[str, Any]]:
"""Get basic channel information from the layout."""
channel_seo = self._normalize_channel_identifier(channel_id)
try:
layout = self.fetch_channel_layout(channel_seo)
entity = layout.get("entity", {})
metadata = entity.get("metadata", {})
return {
"id": entity.get("id"),
"type": entity.get("type"),
"title": metadata.get("title"),
"code": metadata.get("code"),
"seo": layout.get("parent", {}).get("seo"),
}
except Exception as e:
logger.error(f"Failed to get channel info for {channel_id}: {e}")
layout = self._provider.fetch_layout(
layout_type="live",
content_id=channel_seo,
location=f"{self.cfg.beta_website}{channel_seo}/live"
)
if not layout:
return None
entity = layout.get("entity", {})
metadata = entity.get("metadata", {})
return {
"id": entity.get("id"),
"type": entity.get("type"),
"title": metadata.get("title"),
"code": metadata.get("code"),
"seo": layout.get("parent", {}).get("seo"),
}
def get_current_program(self, channel_id: str) -> Optional[Dict[str, Any]]:
"""Get currently playing program info for a channel."""
channel_seo = self._normalize_channel_identifier(channel_id)
try:
layout = self.fetch_channel_layout(channel_seo)
blocks = layout.get("blocks", [])
for block in blocks:
if block.get("type") == "bffPaginated":
items = block.get("content", {}).get("items", [])
for item in items:
if item.get("itemType") == "video":
content = item.get("itemContent", {})
progress = content.get("progressBar", {})
video = content.get("video", {})
progress_data = video.get("progress", {})
layout = self._provider.fetch_layout(
layout_type="live",
content_id=channel_seo,
location=f"{self.cfg.beta_website}{channel_seo}/live"
)
return {
"title": content.get("title"),
"episode_title": content.get("extraTitle"),
"description": content.get("description"),
"progress_percent": progress.get("progressValue"),
"start_time": progress_data.get("startTitle"),
"end_time": progress_data.get("endTitle"),
"live": progress_data.get("live", {}),
}
if not layout:
return None
except Exception as e:
logger.error(f"Failed to get current program for {channel_id}: {e}")
return None
blocks = layout.get("blocks", [])
for block in blocks:
if block.get("type") == "bffPaginated":
items = block.get("content", {}).get("items", [])
for item in items:
if item.get("itemType") == "video":
content = item.get("itemContent", {})
progress = content.get("progressBar", {})
video = content.get("video", {})
progress_data = video.get("progress", {})
return {
"title": content.get("title"),
"episode_title": content.get("extraTitle"),
"description": content.get("description"),
"progress_percent": progress.get("progressValue"),
"start_time": progress_data.get("startTitle"),
"end_time": progress_data.get("endTitle"),
"live": progress_data.get("live", {}),
}
return None
lib/streaming_providers/providers/rtlplus/constants.py
+43 -2
View File
@@ -58,6 +58,7 @@ class RTLPlusDefaults:
TIME_ENDPOINT = "https://time.rtlde.bedrock.tech/"
USERS_ENDPOINT = "https://users.rtlde.bedrock.tech"
PROFILES_PATH = "/v2/platforms/m6group_web/users/{user_id}/profiles"
VERSION_ENDPOINT = "https://beta.plus.rtl.de/version.json"
# DRM license server
DRMTODAY_LICENSE_URL = "https://lic.drmtoday.com/license-proxy-widevine/cenc/"
@@ -99,6 +100,21 @@ class RTLPlusDefaults:
LINEAR_TV_PREFERRED_FORMATS = ["dashcenc", "hlsfp"]
LINEAR_TV_PREFERRED_DRM_TYPES = ["hardware", "software"]
# Layout path patterns
LAYOUT_PATH_LIVE = "/live/{id}/layout"
LAYOUT_PATH_VIDEO = "/video/{id}/layout"
LAYOUT_PATH_FOLDER = "/folder/{id}/layout"
LAYOUT_PATH_PROGRAM = "/program/{id}/layout"
LAYOUT_PATH_BLOCK = "/block/{id}"
# Default pagination values
DEFAULT_BLOCK_PAGE = 1
DEFAULT_NB_PAGES = 2
DEFAULT_BLOCK_NB_PAGES = 3
# Cache TTL for layouts (5 minutes)
LAYOUT_CACHE_TTL = 300
# ---------------------------------------------------------------------------
# VOD — stream config endpoints
# ---------------------------------------------------------------------------
@@ -152,11 +168,11 @@ _CLIENT_VERSION_CACHE: Optional[str] = None
def _get_rtlplus_client_version() -> str:
"""Lazy fetch RTL+ client version from config endpoint."""
"""Lazy fetch RTL+ client version from version.json endpoint."""
global _CLIENT_VERSION_CACHE
if _CLIENT_VERSION_CACHE is None:
try:
with urllib.request.urlopen(RTLPlusDefaults.CONFIG_ENDPOINT, timeout=4) as resp:
with urllib.request.urlopen(RTLPlusDefaults.VERSION_ENDPOINT, timeout=4) as resp:
data = json.loads(resp.read().decode())
_CLIENT_VERSION_CACHE = data["version"]
logger.debug(f"Fetched RTL+ client version: {_CLIENT_VERSION_CACHE}")
@@ -603,3 +619,28 @@ class RTLPlusConfig:
access_token=access_token,
device_id=self.device_id,
)
def get_layout_url(self, layout_type: str, content_id: str) -> str:
"""Get layout URL by type: 'live', 'video', 'folder', 'program', 'block'"""
path_map = {
"live": RTLPlusDefaults.LAYOUT_PATH_LIVE,
"video": RTLPlusDefaults.LAYOUT_PATH_VIDEO,
"folder": RTLPlusDefaults.LAYOUT_PATH_FOLDER,
"program": RTLPlusDefaults.LAYOUT_PATH_PROGRAM,
"block": RTLPlusDefaults.LAYOUT_PATH_BLOCK,
}
path = path_map.get(layout_type)
if not path:
raise ValueError(f"Unknown layout type: {layout_type}")
return f"{self.bedrock_layout_base}{path.format(id=content_id)}"
def get_layout_headers(self, oauth_token: str, bedrock_token: str, location: str = None) -> dict:
"""Common headers for all layout requests"""
return RTLPlusHeaders.get_bedrock_layout_headers(
oauth_token=oauth_token,
bedrock_token=bedrock_token,
device_id=self.device_id,
client_version=self.client_version,
user_agent=self.user_agent,
location=location,
)
lib/streaming_providers/providers/rtlplus/event_manager.py
+482
View File
@@ -0,0 +1,482 @@
# streaming_providers/providers/rtlplus/event_manager.py
"""
RTL+ Event Manager
Handles fetching future/live events from Bedrock layout API.
Replaces legacy GraphQL-based event fetching.
"""
import json
import re
from datetime import datetime
from typing import List, Optional, Dict
from ...base.models.event import Event, EventStatus
from ...base.utils.logger import logger
from ...base.models import DRMConfig
class RTLPlusEventManager:
"""
Manages fetching events (live streams, upcoming sports) for RTL+.
Uses the Bedrock layout API to fetch folder pages and paginated blocks.
"""
# Block titles that indicate live stream content
LIVE_STREAM_BLOCK_TITLES = [
"Sport im Live-Stream",
"UEFA Europa & Conference League | Live",
"Live-Stream",
"Live Events",
]
# Date parsing patterns (German)
DATE_PATTERNS = [
r"(\d{2})\.(\d{2})\.(\d{2}),\s*(\d{2}):(\d{2})\s*Uhr",
r"(\d{2})\.(\d{2})\.(\d{2})\s+(\d{2}):(\d{2})",
r"(\d{2})\.(\d{2})\.(\d{2})", # Just date, no time
]
# Sport type mapping
SPORT_TYPES = {
"Fußball": "Football",
"Motorsport": "Motorsports",
"MMA": "MMA",
"NFL": "American Football",
"Europa League": "Football",
"Conference League": "Football",
"Bundesliga": "Football",
}
def __init__(self, provider):
self._provider = provider
@property
def cfg(self):
return self._provider.rtl_config
@property
def http(self):
return self._provider.http_manager
@property
def auth(self):
return self._provider.authenticator
# --------------------------------------------------------------------------
# Public API
# --------------------------------------------------------------------------
def get_events(
self,
folder_id: str = "6",
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
force_refresh: bool = False,
) -> List[Event]:
"""
Fetch all future/live events from the specified folder.
Args:
folder_id: The folder ID (default "6" for Sport im Überblick)
start_time: Filter events that end after this time
end_time: Filter events that start before this time
force_refresh: Ignore cache and fetch fresh data
Returns:
List of Event objects (scheduled or live)
"""
all_events: List[Event] = []
# Step 1: Fetch folder layout using provider's common method
layout = self._fetch_folder_layout(folder_id, force_refresh)
if not layout:
logger.error(f"Failed to fetch folder layout for {folder_id}")
return []
# Step 2: Identify live stream blocks
live_blocks = self._identify_live_stream_blocks(layout)
if not live_blocks:
logger.debug(f"No live stream blocks found in folder {folder_id}")
return []
logger.info(f"Found {len(live_blocks)} live stream blocks")
# Step 3: Fetch all pages from each block
for block in live_blocks:
block_events = self._fetch_all_block_pages(block, force_refresh)
all_events.extend(block_events)
# Step 4: Filter by time range
filtered_events = self._filter_by_time(all_events, start_time, end_time)
# Step 5: Sort chronologically
filtered_events.sort(key=lambda e: e.start_time if e.start_time else datetime.max)
logger.info(f"Returning {len(filtered_events)} future events from {len(all_events)} total")
return filtered_events
def get_manifest_for_event(self, event_id: str) -> Optional[str]:
"""
Get manifest URL for an event.
Args:
event_id: The event folder ID (e.g., "76")
Returns:
Manifest URL if available from the folder layout, None otherwise
"""
layout = self._fetch_folder_layout(event_id)
if not layout:
return None
# Extract video assets from the layout
assets = self._provider.extract_video_assets(layout)
if not assets:
logger.debug(f"No video assets found in event folder {event_id}")
return None
# Extract best manifest URL
manifest_url = self._provider.extract_best_manifest_url(assets)
if manifest_url:
logger.debug(f"Found manifest for event {event_id}: {manifest_url}")
return self._provider.resolve_redirect(manifest_url)
return None
def get_drm_for_event(self, event_id: str) -> List[DRMConfig]:
"""
Get DRM configuration for an event.
"""
layout = self._fetch_folder_layout(event_id)
if not layout:
return []
# Delegate to provider's common DRM method
return self._provider.get_drm_for_content(layout)
# --------------------------------------------------------------------------
# Internal Methods
# --------------------------------------------------------------------------
def _fetch_folder_layout(self, folder_id: str, force_refresh: bool = False) -> Optional[Dict]:
"""Fetch folder layout using provider's common method."""
return self._provider.fetch_layout(
layout_type="folder",
content_id=folder_id,
force_refresh=force_refresh,
)
def _fetch_block_page(self, block_id: str, page: int, nb_pages: int = 3) -> Optional[Dict]:
"""Fetch a specific page of a block using provider's common method."""
return self._provider.fetch_layout(
layout_type="block",
content_id=block_id,
block_page=page,
nb_pages=nb_pages,
)
def _identify_live_stream_blocks(self, layout: Dict) -> List[Dict]:
"""
Identify blocks that contain live stream events.
Returns:
List of block dictionaries that likely contain future events
"""
blocks = layout.get("blocks", [])
live_blocks = []
for block in blocks:
block_type = block.get("type")
if block_type != "bffPaginated":
continue
block_title = block.get("analytics", {}).get("tealium", {}).get("block_title", "")
# Check if block title matches live stream patterns
is_live_block = any(
title_pattern.lower() in block_title.lower()
for title_pattern in self.LIVE_STREAM_BLOCK_TITLES
)
if is_live_block:
logger.debug(f"Identified live block: '{block_title}'")
live_blocks.append(block)
continue
# Alternative: Check if ANY item in the block has a future date
items = block.get("content", {}).get("items", [])
for item in items:
highlight = item.get("itemContent", {}).get("highlight", "")
if highlight and self._is_future_date(highlight):
logger.debug(f"Block '{block_title}' contains future dates, including")
live_blocks.append(block)
break
return live_blocks
def _fetch_all_block_pages(self, block: Dict, force_refresh: bool = False) -> List[Event]:
"""
Fetch all pages of a paginated block and extract events.
"""
all_events: List[Event] = []
block_id = block.get("id")
if not block_id:
logger.warning("Block missing 'id' field")
return []
# Get pagination info from initial block data
pagination = block.get("content", {}).get("pagination", {})
total_items = pagination.get("totalItems", 0)
items_per_page = pagination.get("itemsPerPage", 4)
next_page = pagination.get("nextPage")
logger.debug(f"Block {block_id}: total={total_items}, per_page={items_per_page}, next_page={next_page}")
# Process the initial page
initial_items = block.get("content", {}).get("items", [])
current_page_events = self._extract_events_from_items(initial_items)
all_events.extend(current_page_events)
# Determine current page based on items count
current_page = 1
if total_items > 0 and items_per_page > 0:
# Rough estimate: if we have more items than items_per_page, we might be on page 2
if len(initial_items) > items_per_page:
current_page = 2
# Fetch remaining pages
while next_page and next_page > current_page:
current_page = next_page
page_data = self._fetch_block_page(block_id, next_page, force_refresh)
if not page_data:
logger.warning(f"Failed to fetch page {next_page} for block {block_id}")
break
items = page_data.get("content", {}).get("items", [])
page_events = self._extract_events_from_items(items)
all_events.extend(page_events)
# Get next_page from the response
pagination = page_data.get("content", {}).get("pagination", {})
next_page = pagination.get("nextPage")
# Safety: prevent infinite loops
if current_page > 50:
logger.warning(f"Reached page limit (50) for block {block_id}")
break
logger.debug(f"Block {block_id}: fetched {len(all_events)} events from {current_page} pages")
return all_events
def _extract_events_from_items(self, items: List[Dict]) -> List[Event]:
"""
Extract Event objects from block items.
"""
events: List[Event] = []
current_time = datetime.now()
for item in items:
if item.get("itemType") != "classic":
continue
item_content = item.get("itemContent", {})
highlight = item_content.get("highlight", "")
if not highlight:
continue
# Parse the date from highlight
event_date = self._parse_date_from_highlight(highlight)
if not event_date:
# Some items might have date in description
description = item_content.get("description", "")
if description:
event_date = self._parse_date_from_highlight(description)
if not event_date:
continue
# Only include future events (or currently live)
if event_date < current_time:
continue
# Extract event data - use itemContent.id which is the content_id for manifest
content_id = item_content.get("id")
title = item_content.get("title")
if not content_id:
continue
if not title:
# Try to extract title from highlight
title = self._extract_title_from_highlight(highlight)
if not title:
title = "Unknown Event"
# Determine status (live if within next 3 hours, else scheduled)
status = EventStatus.LIVE if self._is_currently_live(event_date) else EventStatus.SCHEDULED
# Extract sport type
sport = self._extract_sport_type(highlight)
# Get action target (for potential folder navigation)
action = item_content.get("action", {})
target = action.get("target", {})
# Build event
event = Event(
name=title,
content_id=content_id,
provider=self._provider.provider_name,
start_time=event_date,
end_time=None, # End time not provided in this response
status=status,
logo_url=self._extract_image_url(item_content),
genre=sport,
venue=item_content.get("details") or item_content.get("extraDetails"),
description=item_content.get("description"),
)
# Store additional metadata in manifest_script for later use
manifest_data = {
"folder_id": target.get("value_layout", {}).get("id"),
"seo": target.get("value_layout", {}).get("seo"),
"highlight": highlight,
"title": title,
}
event.manifest_script = json.dumps(manifest_data)
events.append(event)
return events
# --------------------------------------------------------------------------
# Date Parsing Helpers
# --------------------------------------------------------------------------
def _parse_date_from_highlight(self, highlight: str) -> Optional[datetime]:
"""
Parse German date format from highlight string.
Examples:
"Fußball • Sa., 02.05.26, 20:00 Uhr"
"Motorsport • Fr., 15.05.26, 13:10 Uhr"
"Motorsport \u2022 Sa., 16.05.26, 14:15 Uhr"
"Motorsport • Do., 14.05.26, 13:10 Uhr"
"""
for pattern in self.DATE_PATTERNS:
match = re.search(pattern, highlight)
if match:
groups = match.groups()
if len(groups) >= 3:
day = int(groups[0])
month = int(groups[1])
year = int(groups[2])
hour = int(groups[3]) if len(groups) > 3 else 0
minute = int(groups[4]) if len(groups) > 4 else 0
# Assume 20XX for years like '26'
if year < 100:
year = 2000 + year
try:
return datetime(year, month, day, hour, minute)
except ValueError:
continue
return None
def _is_future_date(self, highlight: str) -> bool:
"""Check if the highlight contains a future date."""
event_date = self._parse_date_from_highlight(highlight)
if not event_date:
return False
return event_date > datetime.now()
@staticmethod
def _is_currently_live(event_date: datetime) -> bool:
"""Check if an event is currently live (within 3 hours of start)."""
now = datetime.now()
# Consider event "live" if it started within the last 3 hours
# and hasn't been passed by more than 30 minutes
return event_date <= now <= (event_date.replace(hour=event_date.hour + 3))
@staticmethod
def _extract_title_from_highlight(highlight: str) -> str:
"""Extract event title from highlight text when no separate title exists."""
# Remove sport prefix and date
parts = highlight.split("•")
if len(parts) >= 2:
# Everything between the first bullet and the date
middle = parts[1].strip()
# Remove date part
date_match = re.search(r"\d{2}\.\d{2}\.\d{2}", middle)
if date_match:
title = middle[:date_match.start()].strip()
if title:
return title
return highlight
def _extract_sport_type(self, highlight: str) -> str:
"""Extract sport type from highlight."""
sport_part = highlight.split("•")[0].strip()
for german, english in self.SPORT_TYPES.items():
if german in sport_part:
return english
return sport_part or "Sport"
@staticmethod
def _extract_image_url(item_content: Dict) -> Optional[str]:
"""Extract image URL from item content."""
image = item_content.get("image", {})
if not image:
return None
# Try to get the best available ratio
ratio_prefs = ["16:9", "3:1", "1:1"]
for ratio in ratio_prefs:
image_id = image.get("idsByRatio", {}).get(ratio)
if image_id:
return f"https://images.rtl.de/{image_id}?format=webp&width=400"
# Fallback to direct ID
image_id = image.get("id")
if image_id:
return f"https://images.rtl.de/{image_id}?format=webp&width=400"
return None
@staticmethod
def _filter_by_time(
events: List[Event],
start_time: Optional[datetime],
end_time: Optional[datetime],
) -> List[Event]:
"""Filter events by time range."""
if not start_time and not end_time:
return events
filtered = []
for event in events:
if start_time and event.end_time and event.end_time < start_time:
continue
if end_time and event.start_time and event.start_time > end_time:
continue
filtered.append(event)
return filtered
# --------------------------------------------------------------------------
# Cache Management
# --------------------------------------------------------------------------
def invalidate_cache(self, folder_id: Optional[str] = None):
"""Invalidate the layout cache."""
if folder_id:
cache_key = f"folder:{folder_id}:1:2" # Match _fetch_layout cache key format
self._provider.invalidate_layout_cache(cache_key)
else:
self._provider.invalidate_layout_cache()
lib/streaming_providers/providers/rtlplus/provider.py
+439 -93
View File
@@ -2,19 +2,21 @@
import json
import time
from datetime import datetime
from typing import ClassVar, Dict, List, Optional, Any
from typing import ClassVar, Dict, List, Optional, Tuple
import urllib.parse
import requests
from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel, Event
from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel, Event, LicenseUnwrapperParams
from ...base.models.proxy_models import ProxyConfig
from ...base.provider import StreamingProvider
from ...base.utils import logger
from .auth import RTLPlusAuthenticator
from .constants import RTLPlusConfig, RTLPlusDefaults, RTLPlusGraphQL
from .models import RTLPlusLiveEvent
from .constants import RTLPlusConfig, RTLPlusDefaults
from .vod_manager import RTLPlusVodManager
from .channel_manager import RTLPlusChannelManager
from .event_manager import RTLPlusEventManager
_MANIFEST_CACHE_TTL = 86400 # 1 day in seconds for VOD/events
@@ -57,6 +59,10 @@ class RTLPlusProvider(StreamingProvider):
self.http_manager = self._share_http_manager_with_authenticator(self.authenticator)
# Layout cache (shared across all layout types)
self._layout_cache: Dict[str, Tuple[Dict, float]] = {}
self.LAYOUT_CACHE_TTL = RTLPlusDefaults.LAYOUT_CACHE_TTL
# Try authentication
try:
self.bearer_token = self.authenticator.get_bearer_token()
@@ -73,6 +79,7 @@ class RTLPlusProvider(StreamingProvider):
# Initialize managers
self._vod_manager = RTLPlusVodManager(self)
self.channel_manager = RTLPlusChannelManager(self)
self.event_manager = RTLPlusEventManager(self)
# Manifest cache for VOD/events
self._manifest_cache: Dict[str, tuple] = {}
@@ -101,6 +108,235 @@ class RTLPlusProvider(StreamingProvider):
def supported_auth_types(self) -> List[str]:
return ["user_credentials"]
# --------------------------------------------------------------------------
# Common Layout Methods
# --------------------------------------------------------------------------
def fetch_layout(
self,
layout_type: str,
content_id: str,
block_page: int = None,
nb_pages: int = None,
location: str = None,
force_refresh: bool = False,
) -> Optional[Dict]:
"""
Fetch any layout (live/video/folder/program/block) with caching.
Args:
layout_type: 'live', 'video', 'folder', 'program', 'block'
content_id: The ID (seo for live, clip_id for video, folder_id for folder)
block_page: Page number for blocks (default 1)
nb_pages: Number of pages to request (default 2)
location: x-location header value (auto-generated if not provided)
force_refresh: Ignore cache
Returns:
Layout JSON or None
"""
if block_page is None:
block_page = RTLPlusDefaults.DEFAULT_BLOCK_PAGE
if nb_pages is None:
nb_pages = RTLPlusDefaults.DEFAULT_NB_PAGES
cache_key = f"{layout_type}:{content_id}:{block_page}:{nb_pages}"
now = time.time()
if not force_refresh and cache_key in self._layout_cache:
cached_data, cached_time = self._layout_cache[cache_key]
if (now - cached_time) < self.LAYOUT_CACHE_TTL:
logger.debug(f"Layout cache hit: {cache_key}")
return cached_data
# Get tokens
oauth_token = self.get_user_bearer_token()
if not oauth_token:
try:
oauth_token = self.authenticator.get_bearer_token()
logger.debug("Using anonymous token for layout")
except Exception as e:
logger.error(f"Failed to get OAuth token for layout: {e}")
return None
try:
bedrock_token = self.authenticator.get_bedrock_token()
except Exception as e:
logger.error(f"Failed to get Bedrock token for layout: {e}")
return None
# Auto-generate location header if not provided
if location is None and layout_type in ("live", "video", "folder", "program"):
location = f"{self.rtl_config.beta_website}{content_id}"
# Build request
url = self.rtl_config.get_layout_url(layout_type, content_id)
headers = self.rtl_config.get_layout_headers(oauth_token, bedrock_token, location)
params = {"blockPage": block_page, "nbPages": nb_pages}
try:
response = self.http_manager.get(
url, headers=headers, params=params, operation="api"
)
response.raise_for_status()
data = response.json()
self._layout_cache[cache_key] = (data, now)
logger.debug(f"Fetched {layout_type} layout for {content_id}")
return data
except Exception as e:
logger.error(f"Failed to fetch {layout_type} layout for {content_id}: {e}")
return None
@staticmethod
def extract_video_assets(layout_data: Dict) -> List[Dict]:
"""
Extract video assets from a layout response.
Returns list of asset dicts with keys: path, quality, format, drm, etc.
"""
assets = []
blocks = layout_data.get("blocks", [])
for block in blocks:
if block.get("type") != "bffPaginated":
continue
items = block.get("content", {}).get("items", [])
for item in items:
if item.get("itemType") == "video":
video = item.get("itemContent", {}).get("video", {})
assets.extend(video.get("assets", []))
elif item.get("itemType") == "classic":
# Some layouts wrap video inside classic items
item_content = item.get("itemContent", {})
video = item_content.get("video", {})
assets.extend(video.get("assets", []))
return assets
@staticmethod
def _extract_items_from_block(
layout_data: Dict,
block_type: str = None,
item_type: str = None
) -> List[Dict]:
"""
Extract items from layout blocks with optional filtering.
Args:
layout_data: Layout JSON
block_type: Filter blocks by type (e.g., 'bffPaginated')
item_type: Filter items by itemType (e.g., 'classic', 'video')
Returns:
List of item dictionaries
"""
items = []
blocks = layout_data.get("blocks", [])
for block in blocks:
if block_type and block.get("type") != block_type:
continue
block_items = block.get("content", {}).get("items", [])
for item in block_items:
if item_type and item.get("itemType") != item_type:
continue
items.append(item)
return items
@staticmethod
def _get_pagination_info(layout_data: Dict, block_id: str = None) -> Dict:
"""
Extract pagination info from layout or specific block.
"""
if block_id:
for block in layout_data.get("blocks", []):
if block.get("id") == block_id or block.get("blockId") == block_id:
return block.get("content", {}).get("pagination", {})
# Fallback: look for pagination at top level
return layout_data.get("pagination", {})
def extract_best_manifest_url(
self,
assets: List[Dict],
preferred_quality: str = None,
preferred_format: str = None,
) -> Optional[str]:
"""
Extract the best manifest URL from assets based on preferences.
"""
quality = preferred_quality or next(iter(self.rtl_config.preferred_qualities), "hd")
format_pref = preferred_format or next(iter(self.rtl_config.preferred_formats), "dashcenc")
# Try exact match
for asset in assets:
if (asset.get("quality") == quality and
asset.get("format") == format_pref):
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
return manifest_url
# Try format + quality, any DRM
for asset in assets:
if asset.get("quality") == quality and asset.get("format") == format_pref:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
return manifest_url
# Try by preferred formats/qualities
for fmt in self.rtl_config.preferred_formats:
for qual in self.rtl_config.preferred_qualities:
for asset in assets:
if asset.get("format") == fmt and asset.get("quality") == qual:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
return manifest_url
# Last resort: any asset with a path
for asset in assets:
manifest_url = asset.get("path") or asset.get("reference")
if manifest_url:
return manifest_url
return None
def resolve_redirect(self, url: str) -> str:
"""
Resolve HTTP redirects to get final manifest URL.
"""
if not url:
return url
try:
response = self.http_manager.head(
url,
headers={"User-Agent": self.rtl_config.user_agent},
follow_redirects=False,
timeout=10
)
if 300 <= response.status_code < 400:
location = response.headers.get("location")
if location:
logger.debug(f"Resolved redirect: {url} -> {location}")
return location
except Exception as e:
logger.debug(f"Redirect resolution failed: {e}")
return url
def invalidate_layout_cache(self, cache_key: str = None):
"""Invalidate layout cache for a specific key or all keys."""
if cache_key:
self._layout_cache.pop(cache_key, None)
else:
self._layout_cache.clear()
# --------------------------------------------------------------------------
# Authentication Helpers
# --------------------------------------------------------------------------
@@ -111,8 +347,8 @@ class RTLPlusProvider(StreamingProvider):
try:
current_level = self.authenticator.get_current_token_level()
force_upgrade = (
self.authenticator.has_user_credentials()
and current_level != TokenAuthLevel.USER_AUTHENTICATED
self.authenticator.has_user_credentials()
and current_level != TokenAuthLevel.USER_AUTHENTICATED
)
bearer_token = self.authenticator.get_bearer_token(force_upgrade=force_upgrade)
except Exception as e:
@@ -138,76 +374,25 @@ class RTLPlusProvider(StreamingProvider):
return []
# --------------------------------------------------------------------------
# Events (Live Events)
# Events
# --------------------------------------------------------------------------
def get_events(
self,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**kwargs,
self,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**kwargs,
) -> List[Event]:
"""Fetch upcoming / live RTL+ events from the editorial GraphQL endpoint."""
raw_events = self._fetch_live_events()
events: List[Event] = []
for raw in raw_events:
try:
event = raw.to_event(provider=self.provider_name)
except Exception as e:
logger.warning(f"RTL+: Could not convert event '{raw.id}': {e}")
continue
if start_time and event.end_time and event.end_time < start_time:
continue
if end_time and event.start_time and event.start_time > end_time:
continue
events.append(event)
logger.info(f"RTL+: Fetched {len(events)} events")
return events
def _fetch_live_events(self) -> List[RTLPlusLiveEvent]:
headers = self._get_rtlplus_authenticated_headers()
try:
response = self.http_manager.get(
self.rtl_config.graphql_endpoint,
params=RTLPlusGraphQL.live_events_overview_page(),
headers=headers,
operation="api",
)
response.raise_for_status()
except Exception as e:
logger.error(f"RTL+: Failed to fetch events: {e}")
return []
return self._parse_live_events(response.json())
@staticmethod
def _parse_live_events(data: Dict[str, Any]) -> List[RTLPlusLiveEvent]:
events: List[RTLPlusLiveEvent] = []
seen: set = set()
teaser_rows = data.get("data", {}).get("liveEventsOverview", {}).get("teaserRows", [])
for row in teaser_rows:
for element in (row.get("events") or []):
if element is None:
continue
if element.get("__typename") != "LiveEvent":
continue
event_id = element.get("id", "")
if not event_id or event_id in seen:
continue
seen.add(event_id)
try:
events.append(RTLPlusLiveEvent.from_api_node(element))
except Exception as e:
logger.warning(f"RTL+: Skipping malformed event node: {e}")
return events
"""
Fetch upcoming / live RTL+ events from the Bedrock layout API.
"""
folder_id = kwargs.get("folder_id", "6") # Default to Sport folder
return self.event_manager.get_events(
folder_id=folder_id,
start_time=start_time,
end_time=end_time,
force_refresh=kwargs.get("force_refresh", False),
)
# --------------------------------------------------------------------------
# VOD
@@ -217,34 +402,44 @@ class RTLPlusProvider(StreamingProvider):
return self._vod_manager.get_vod_category(content_id=content_id, **kwargs)
# --------------------------------------------------------------------------
# Manifest & DRM (Dispatcher)
# Manifest & DRM (Unified)
# --------------------------------------------------------------------------
@staticmethod
def _is_linear_tv_channel(content_id: str) -> bool:
"""Determine if content_id refers to a linear TV channel."""
return (
":" not in content_id
and not content_id.startswith("rrn:")
and not content_id.startswith("/")
and not content_id.startswith("http")
)
def get_manifest(self, content_id: str, **kwargs) -> Optional[str]:
"""
Get manifest URL for content.
For linear TV channels: uses new Bedrock layout API
For VOD: uses existing GraphQL/Wurstland flow
For live events: uses existing event manifest flow
Supports:
- Linear TV channels (via channel_manager)
- VOD clips (via layout extraction)
- Events (via event_manager)
"""
# Try linear TV channel first
if self._is_linear_tv_channel(content_id):
return self.channel_manager.get_best_manifest_url(content_id)
else:
return self._get_manifest_vod_or_event(content_id, **kwargs)
# Try as event (folder) - event_manager will return manifest if available
if content_id.isdigit() and int(content_id) > 0:
manifest = self.event_manager.get_manifest_for_event(content_id)
if manifest:
return manifest
# Fall back to VOD/event manifest extraction
return self._get_manifest_vod_or_event(content_id, **kwargs)
@staticmethod
def _is_linear_tv_channel(content_id: str) -> bool:
"""Determine if content_id refers to a linear TV channel."""
return (
":" not in content_id
and not content_id.startswith("rrn:")
and not content_id.startswith("/")
and not content_id.startswith("http")
and not content_id.isdigit() # Event folder IDs are digits
)
def _get_manifest_vod_or_event(self, content_id: str, **kwargs) -> Optional[str]:
"""Original manifest logic for VOD/events."""
"""Original manifest logic for VOD/events using layout extraction."""
try:
manifest_data = self._fetch_manifest_data(content_id)
if manifest_data is None:
@@ -283,17 +478,168 @@ class RTLPlusProvider(StreamingProvider):
logger.error(f"RTL+ Manifest Unexpected Error: {str(e)}")
return None
# Add to RTLPlusProvider class
def get_drm_for_content(self, layout_data: Dict) -> List[DRMConfig]:
"""
Extract DRM configuration from layout data.
This method works for any layout type (live channel, event folder, VOD)
that contains video assets with DRM configuration.
Args:
layout_data: Layout JSON from fetch_layout()
Returns:
List of DRMConfig objects
"""
assets = self.extract_video_assets(layout_data)
if not assets:
logger.debug("No video assets found in layout")
return []
# Priority: delta provider dashcenc format (best quality)
target_asset = None
# First try: delta + dashcenc + hd quality
for asset in assets:
if (asset.get("provider") == "delta" and
asset.get("format") == "dashcenc" and
asset.get("quality") == "hd"):
target_asset = asset
logger.debug("Found delta/dashcenc/hd asset")
break
# Fallback: delta + dashcenc + any quality
if not target_asset:
for asset in assets:
if asset.get("provider") == "delta" and asset.get("format") == "dashcenc":
target_asset = asset
logger.debug(f"Fallback to delta/dashcenc/{asset.get('quality', 'unknown')} asset")
break
# Last resort: any dashcenc asset
if not target_asset:
for asset in assets:
if asset.get("format") == "dashcenc":
target_asset = asset
logger.debug(f"Last resort: {asset.get('provider', 'unknown')}/dashcenc asset")
break
if not target_asset:
logger.warning("No suitable DRM asset found")
return []
# Extract contentId from the selected asset
drm_info = target_asset.get("drm", {})
drm_config = drm_info.get("config", {})
content_id = drm_config.get("contentId")
if not content_id:
logger.error("No contentId in asset DRM config")
return []
try:
uid = self.authenticator.get_user_id_from_token()
if not uid:
logger.error("No user ID available for DRM")
return []
# Get upfront token (works for both Widevine and PlayReady)
upfront_token = self.authenticator.get_upfront_token(
content_id=content_id,
uid=uid,
)
if not upfront_token:
logger.error(f"Failed to get upfront token for {content_id}")
return []
drm_configs = []
# Build Widevine config
wv_config = self._get_widevine_config(upfront_token)
if wv_config:
drm_configs.append(wv_config)
logger.debug("Added Widevine DRM")
# Build PlayReady config
pr_config = self._get_playready_config(upfront_token)
if pr_config:
drm_configs.append(pr_config)
logger.debug("Added PlayReady DRM")
logger.info(f"Built {len(drm_configs)} DRM configs (Widevine + PlayReady)")
return drm_configs
except Exception as e:
logger.error(f"Failed to get DRM: {e}")
return []
def _get_widevine_config(self, upfront_token: str) -> Optional[DRMConfig]:
"""Get Widevine DRM configuration."""
if not upfront_token:
return None
headers = self.rtl_config.get_drm_license_headers(upfront_token)
req_headers = urllib.parse.urlencode(headers)
license_config = LicenseConfig(
server_url=self.rtl_config.drmtoday_license_url,
req_headers=req_headers,
req_data="{CHA-RAW}",
use_http_get_request=False,
unwrapper="json,base64",
unwrapper_params=LicenseUnwrapperParams(path_data="license"),
)
return DRMConfig(
system=DRMSystem.WIDEVINE,
priority=3,
license=license_config,
)
def _get_playready_config(self, upfront_token: str) -> Optional[DRMConfig]:
"""Get PlayReady DRM configuration."""
if not upfront_token:
return None
headers = self.rtl_config.get_playready_license_headers(upfront_token)
req_headers = urllib.parse.urlencode(headers)
return DRMConfig(
system=DRMSystem.PLAYREADY,
priority=2,
license=LicenseConfig(
server_url=RTLPlusDefaults.DRMTODAY_PLAYREADY_URL,
req_headers=req_headers,
req_data="{CHA-RAW}",
use_http_get_request=False,
),
)
def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
"""
Get DRM configuration for content.
For linear TV channels: uses new upfront token flow
For VOD: uses existing manifest-based DRM
Supports:
- Linear TV channels (via channel_manager)
- VOD clips (via manifest extraction)
- Events (via event_manager)
"""
# Try linear TV channel first
if self._is_linear_tv_channel(content_id):
return self.channel_manager.get_drm_config_for_channel(content_id)
else:
return self._get_drm_vod_or_event(content_id, **kwargs)
# Try as event (folder)
if content_id.isdigit() and int(content_id) > 0:
drm_configs = self.event_manager.get_drm_for_event(content_id)
if drm_configs:
return drm_configs
# Fall back to VOD/event DRM extraction
return self._get_drm_vod_or_event(content_id, **kwargs)
def _get_drm_vod_or_event(self, content_id: str, **kwargs) -> List[DRMConfig]:
"""Original DRM logic for VOD/events."""