nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-12 10:52:23 +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-23 20:55:52 +02:00
parent 780856897d
commit 26d34dcb84
5 changed files with 873 additions and 703 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/rtlplus/auth.py
+154 -147
View File
@@ -1,6 +1,7 @@
# streaming_providers/providers/rtlplus/auth.py
import base64
import json
import time
from typing import Any, Dict, Optional
from ...base.auth.base_auth import BaseAuthToken, TokenAuthLevel
@@ -13,16 +14,15 @@ from .models import RTLPlusAuthToken, RTLPlusClientCredentials, RTLPlusUserCrede
class RTLPlusAuthenticator(BaseOAuth2Authenticator):
def __init__(
self,
credentials=None,
config_dir=None,
client_version=None,
device_id=None,
proxy_config: Optional[ProxyConfig] = None,
http_manager=None,
self,
credentials=None,
config_dir=None,
client_version=None,
device_id=None,
proxy_config: Optional[ProxyConfig] = None,
http_manager=None,
):
# Initialize configuration FIRST
config_dict = {}
if client_version:
config_dict["client_version"] = client_version
@@ -31,33 +31,31 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
self._config = RTLPlusConfig(config_dict)
self._client_id = None
self._bedrock_token: Optional[str] = None
self._bedrock_token_expiry: float = 0
self._cached_user_id: Optional[str] = None
# Get proxy_config if not provided
if proxy_config is None:
from ...base.network import ProxyConfigManager
proxy_mgr = ProxyConfigManager(config_dir)
proxy_config = proxy_mgr.get_proxy_config("rtlplus")
# Call parent init FIRST
super().__init__(
provider_name="rtlplus",
credentials=credentials, # Pass None if not provided
credentials=credentials,
config_dir=config_dir,
proxy_config=proxy_config,
http_manager=http_manager,
)
# NOW set default credentials if needed (after super init)
if self.credentials is None:
self.credentials = self._get_default_credentials()
@property
def auth_endpoint(self) -> str:
"""Override auth_endpoint to use our config"""
return self.config.auth_endpoint
# Required OAuth2 properties
@property
def oauth_client_id(self) -> str:
return self._get_client_id()
@@ -70,36 +68,30 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
def oauth_redirect_uri(self) -> str:
return self.config.base_website
@property
def config(self) -> RTLPlusConfig:
return self._config
def _get_auth_headers(self) -> Dict[str, str]:
"""RTL+-specific authentication headers"""
return self.config.get_auth_headers()
def _build_auth_payload(self) -> Dict[str, Any]:
"""Build authentication payload from credentials"""
return self.credentials.to_auth_payload()
def _get_default_credentials(self):
"""Get default client credentials for anonymous access"""
try:
# Try to get dynamic credentials first
config_creds = self._get_anonymous_credentials_from_config()
if config_creds:
return RTLPlusClientCredentials(
client_id=config_creds.get("client_id", RTLPlusDefaults.ANONYMOUS_CLIENT_ID),
client_secret=config_creds.get(
"client_secret", RTLPlusDefaults.ANONYMOUS_CLIENT_SECRET
),
client_secret=config_creds.get("client_secret", RTLPlusDefaults.ANONYMOUS_CLIENT_SECRET),
)
except Exception as e:
logger.warning(f"Could not get dynamic credentials: {e}")
# Fallback to default credentials
return RTLPlusClientCredentials()
def _create_token_from_response(self, response_data: Dict[str, Any]) -> RTLPlusAuthToken:
"""Create RTL+-specific token from OAuth2 response"""
import time
return RTLPlusAuthToken(
access_token=response_data["access_token"],
token_type=response_data.get("token_type", "Bearer"),
@@ -111,14 +103,7 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
scope=response_data.get("scope", ""),
)
def get_current_token_level(self) -> "TokenAuthLevel":
"""
Return the auth level of whichever token the base class currently holds.
Walks the common attribute names used by BaseOAuth2Authenticator to
store the active token, classifies it via _classify_token, and returns
UNKNOWN if no token is found.
"""
def get_current_token_level(self) -> TokenAuthLevel:
for attr in ("token", "_token", "current_token", "_current_token", "_access_token"):
tok = getattr(self, attr, None)
if tok is not None:
@@ -126,36 +111,17 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
return TokenAuthLevel.UNKNOWN
def get_fallback_credentials(self):
"""Get fallback credentials (anonymous client credentials)"""
return self._get_default_credentials()
def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel:
"""
Classify RTL+ token authentication level by decoding JWT payload
Logic:
- CLIENT_CREDENTIALS: isGuest=True AND clientId='anonymous-user'
- USER_AUTHENTICATED: Has preferred_username OR email claims
- UNKNOWN: Cannot determine or invalid token
Args:
token: Token to classify
Returns:
TokenAuthLevel indicating the authentication level
"""
if not token or not token.access_token:
logger.debug("RTL+ Cannot classify: No token or access token")
return TokenAuthLevel.UNKNOWN
try:
# Decode JWT without verification to check the payload
parts = token.access_token.split(".")
if len(parts) < 2:
logger.debug("RTL+ Cannot classify: Invalid token format")
return TokenAuthLevel.UNKNOWN
# Add padding if needed and decode
payload_segment = parts[1]
padding = 4 - len(payload_segment) % 4
if padding != 4:
@@ -164,61 +130,38 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
payload_json = base64.b64decode(payload_segment)
payload = json.loads(payload_json)
# Extract relevant claims
client_id = payload.get("clientId")
is_guest = payload.get("isGuest", False)
preferred_username = payload.get("preferred_username")
email = payload.get("email")
logger.debug(
f"RTL+ Token JWT payload: clientId={client_id}, isGuest={is_guest}, "
f"has_preferred_username={bool(preferred_username)}, has_email={bool(email)}"
)
# Check for user-authenticated token
if preferred_username or email:
logger.debug("RTL+ Token classified as USER_AUTHENTICATED (has user claims)")
return TokenAuthLevel.USER_AUTHENTICATED
# Check for client credentials (anonymous) token
if is_guest and client_id == "anonymous-user":
logger.debug("RTL+ Token classified as CLIENT_CREDENTIALS (anonymous)")
return TokenAuthLevel.CLIENT_CREDENTIALS
# Cannot determine
logger.debug("RTL+ Token classified as UNKNOWN (no matching criteria)")
return TokenAuthLevel.UNKNOWN
except Exception as e:
logger.warning(f"RTL+ Error classifying token: {e}")
return TokenAuthLevel.UNKNOWN
def _perform_oauth_authorization_code_flow(
self, username: str, password: str
) -> Dict[str, Any]:
"""
RTL+ specific OAuth2 authorization code flow with PKCE
Uses base class generic form login
"""
def _perform_oauth_authorization_code_flow(self, username: str, password: str) -> Dict[str, Any]:
import uuid
return self._perform_generic_form_login(
username=username,
password=password,
form_selector_pattern=r'<form id="rtlplus-form-login" action="([^"]*)"',
login_fields={"username": "username", "password": "password"},
extra_params={
"prompt": "login",
"nonce": str(uuid.uuid4()),
},
extra_params={"prompt": "login", "nonce": str(uuid.uuid4())},
additional_form_data={"credentialId": "", "rememberMe": "on"},
)
def _get_client_id(self) -> str:
"""Get client ID from RTL+ website configuration using base class method"""
if self._client_id:
return self._client_id
# Use base class method for extraction
self._client_id = self._extract_client_id_from_js(
main_page_url=self.config.base_website,
js_file_pattern=r'<script src="(main[A-z0-9\-\.]+\.js)"',
@@ -228,26 +171,21 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
if self._client_id:
return self._client_id
# Fallback to default if extraction failed
logger.warning("Could not extract client ID, using default")
return RTLPlusDefaults.CLIENT_ID
def _get_client_version(self) -> str:
"""Get client version from RTL+ configuration"""
if self.config.client_version != RTLPlusDefaults.CLIENT_VERSION:
if self.config.client_version != RTLPlusDefaults.CLIENT_VERSION_FALLBACK:
return self.config.client_version
try:
headers = self.config.get_base_headers()
response = self.http_manager.get(
self.config.config_endpoint, operation="api", headers=headers
)
response = self.http_manager.get(self.config.config_endpoint, operation="api", headers=headers)
response.raise_for_status()
config_data = response.json()
version = config_data.get("version", RTLPlusDefaults.CLIENT_VERSION)
version = config_data.get("version", RTLPlusDefaults.CLIENT_VERSION_FALLBACK)
# Update config with retrieved version
self.config.client_version = version
return version
@@ -256,13 +194,7 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
return self.config.client_version
def _get_anonymous_credentials_from_config(self) -> Optional[Dict[str, str]]:
"""
Extract anonymous credentials from RTL+ website configuration
Uses base class generic config extraction
"""
def parse_credentials(config_str: str) -> Dict[str, str]:
"""Parse anonymousCredentials config string"""
credentials = {}
for pair in config_str.split(","):
if ":" in pair:
@@ -279,41 +211,145 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
parse_function=parse_credentials,
)
# RTL+-specific credential management methods
def set_user_credentials(
self, username: str, password: str, client_id: Optional[str] = None
) -> bool:
# --------------------------------------------------------------------------
# Bedrock Token Management (Linear TV)
# --------------------------------------------------------------------------
def get_bedrock_token(self, force_refresh: bool = False) -> str:
"""
Set RTL+ user credentials for authentication
Get or refresh Bedrock token.
The Bedrock token is a JWT required for all Bedrock API calls
(layout, DRM, heartbeat). Obtained using the OAuth token.
"""
if not force_refresh and self._bedrock_token and self._bedrock_token_expiry > time.time() + 300:
return self._bedrock_token
oauth_token = self.get_bearer_token()
response = self.http_manager.post(
self.config.bedrock_auth_url,
headers={"authorization": f"Bearer {oauth_token}"},
operation="api",
)
response.raise_for_status()
data = response.json()
self._bedrock_token = data.get("token")
if self._bedrock_token:
try:
parts = self._bedrock_token.split(".")
if len(parts) >= 2:
payload = parts[1]
padding = 4 - len(payload) % 4
if padding != 4:
payload += "=" * padding
decoded = json.loads(base64.b64decode(payload))
self._bedrock_token_expiry = decoded.get("exp", 0)
except Exception as e:
logger.debug(f"Could not decode Bedrock token expiry: {e}")
logger.debug("RTL+ Bedrock token obtained successfully")
return self._bedrock_token
def get_upfront_token(self, content_id: str, uid: str) -> str:
"""
Get upfront token for DRM license acquisition.
The upfront token is used as x-dt-auth-token when requesting
licenses from DRMToday.
Args:
username: RTL+ username/email
password: RTL+ password
client_id: Optional client ID for user authentication
content_id: DRM content ID (e.g., "dashcenc_rtlde_vox")
uid: User ID from OAuth token
"""
oauth_token = self.get_bearer_token()
bedrock_token = self.get_bedrock_token()
url = self.config.get_upfront_token_url(uid=uid, content_id=content_id)
headers = self.config.get_upfront_token_headers(oauth_token, bedrock_token)
response = self.http_manager.get(url, headers=headers, operation="api")
response.raise_for_status()
data = response.json()
token = data.get("token")
if not token:
raise ValueError("No token in upfront token response")
logger.debug(f"RTL+ Upfront token obtained for {content_id}")
return token
def invalidate_bedrock_token(self) -> None:
"""Invalidate cached Bedrock token"""
self._bedrock_token = None
self._bedrock_token_expiry = 0
def get_user_id_from_token(self) -> Optional[str]:
"""
Extract user ID (sub claim) from the current OAuth token.
Returns:
True if credentials were set and saved successfully
User ID string or None if not available
"""
try:
# Create new user credentials
user_creds = RTLPlusUserCredentials(
username=username, password=password, client_id=client_id
)
if self._cached_user_id:
return self._cached_user_id
token = self.get_bearer_token()
if not token:
return None
try:
parts = token.split(".")
if len(parts) < 2:
return None
payload_segment = parts[1]
padding = 4 - len(payload_segment) % 4
if padding != 4:
payload_segment += "=" * padding
payload_json = base64.b64decode(payload_segment)
payload = json.loads(payload_json)
# The user ID is in the 'sub' claim
# Format: f:83a2e227-f27d-4d33-a811-33ad588170c4:1052940424
sub = payload.get("sub", "")
# Extract numeric ID from the end if present
if ":" in sub:
self._cached_user_id = sub.split(":")[-1]
else:
self._cached_user_id = sub
return self._cached_user_id
except Exception as e:
logger.warning(f"Could not extract user ID from token: {e}")
return None
# --------------------------------------------------------------------------
# Credential Management
# --------------------------------------------------------------------------
def set_user_credentials(self, username: str, password: str, client_id: Optional[str] = None) -> bool:
try:
user_creds = RTLPlusUserCredentials(username=username, password=password, client_id=client_id)
# Validate credentials
if not user_creds.validate():
logger.warning("Invalid user credentials provided")
return False
# Set as current credentials
self.credentials = user_creds
# Save to persistent storage
success = self.save_credentials(user_creds)
if success:
logger.info("RTL+ user credentials saved successfully")
# Invalidate current token to force re-authentication with new credentials
self.invalidate_token()
self.invalidate_bedrock_token()
self._cached_user_id = None # Clear cached user ID
else:
logger.error("Failed to save RTL+ user credentials")
@@ -324,61 +360,32 @@ class RTLPlusAuthenticator(BaseOAuth2Authenticator):
return False
def has_user_credentials(self) -> bool:
"""
Check if user credentials are currently set (not anonymous)
Returns:
True if using user credentials, False if using anonymous access
"""
from ...base.auth.credentials import UserPasswordCredentials
return isinstance(self.credentials, (RTLPlusUserCredentials, UserPasswordCredentials))
def has_stored_credentials(self) -> bool:
"""
Check for stored RTL+ user credentials
"""
try:
logger.debug("RTL+ Checking for stored credentials using settings manager")
stored_creds = self.settings_manager.get_provider_credentials(self.provider_name)
if not stored_creds:
logger.debug("RTL+ No stored credentials found")
return False
# Import the base credential types
from ...base.auth.credentials import UserPasswordCredentials
# Check if it's either RTLPlusUserCredentials OR base UserPasswordCredentials
is_user_creds = isinstance(
stored_creds, (RTLPlusUserCredentials, UserPasswordCredentials)
)
logger.debug(
f"RTL+ Has stored user credentials: {is_user_creds} (type: {type(stored_creds)})"
)
return is_user_creds
return isinstance(stored_creds, (RTLPlusUserCredentials, UserPasswordCredentials))
except Exception as e:
logger.debug(f"RTL+ Error checking stored credentials: {e}")
return False
def get_authentication_status(self) -> Dict[str, Any]:
"""
Get RTL+-specific authentication status information
"""
status = super().get_authentication_status()
# Add RTL+-specific information
status.update(
{
"has_user_credentials": self.has_user_credentials(),
"authentication_mode": ("user" if self.has_user_credentials() else "anonymous"),
"client_version": self.config.client_version,
}
)
status.update({
"has_user_credentials": self.has_user_credentials(),
"authentication_mode": ("user" if self.has_user_credentials() else "anonymous"),
"client_version": self.config.client_version,
"has_bedrock_token": bool(self._bedrock_token),
"has_user_id": bool(self._cached_user_id),
})
if self.has_user_credentials() and hasattr(self.credentials, "username"):
status["username"] = self.credentials.username
return status
lib/streaming_providers/providers/rtlplus/channel_manager.py
+375
View File
@@ -0,0 +1,375 @@
# streaming_providers/providers/rtlplus/channel_manager.py
"""
RTL+ Linear TV Channel Manager
Handles all linear TV (live channel) functionality:
- Fetching channel layouts
- Extracting stream assets
- Manifest URL discovery
- DRM configuration
- Channel listing from EPG grid
"""
from typing import List, Optional, Dict, Any
import json
import time
from datetime import date
from ...base.models import DRMConfig, DRMSystem, LicenseConfig
from ...base.utils.logger import logger
class RTLPlusChannelManager:
"""
Manages linear TV channels for RTL+.
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):
return self._provider.rtl_config
@property
def http(self):
return self._provider.http_manager
@property
def auth(self):
return self._provider.authenticator
# --------------------------------------------------------------------------
# User ID Helpers
# --------------------------------------------------------------------------
def _get_user_id(self) -> str:
"""Get user ID from authenticator."""
uid = self.auth.get_user_id_from_token()
if uid:
return uid
raise ValueError("Could not determine user ID for DRM token request")
# --------------------------------------------------------------------------
# Channel Listing (EPG Grid)
# --------------------------------------------------------------------------
def get_channels(self, nb_pages: int = 3) -> List[Dict[str, Any]]:
"""
Get list of all linear TV channels from the EPG grid endpoint.
Args:
nb_pages: Number of EPG grid pages to fetch (default 3)
Returns:
List of channel dicts with id, title, logo_url, seo, etc.
"""
try:
oauth_token = self.auth.get_bearer_token()
bedrock_token = self.auth.get_bedrock_token()
if not bedrock_token:
logger.error("Failed to obtain Bedrock token")
return []
except Exception as e:
logger.error(f"Failed to obtain tokens for channel listing: {e}")
return []
url = self.cfg.get_epg_grid_url()
location = "https://plus.rtl.de/tv-programm"
headers = self.cfg.get_bedrock_layout_headers(oauth_token, bedrock_token, location)
today = date.today().isoformat()
params = {"day": today, "nbPages": nb_pages}
try:
response = self.http.get(url, headers=headers, params=params, operation="api")
response.raise_for_status()
data = response.json()
except Exception as e:
logger.error(f"Failed to fetch EPG grid: {e}")
return []
channels = []
content = data.get("content", {})
items = content.get("items", [])
for item in items:
if item.get("itemType") == "epg":
item_content = item.get("itemContent", {})
channel_data = item_content.get("channel", {})
action = item_content.get("action", {})
target = action.get("target", {})
value_layout = target.get("value_layout", {})
channel = {
"id": channel_data.get("id"),
"title": channel_data.get("title"),
"seo": value_layout.get("seo"),
"slug": value_layout.get("id"),
"logo_id": channel_data.get("image", {}).get("id"),
"channel_type": "BROADCAST",
}
channels.append(channel)
logger.info(f"Found {len(channels)} channels from EPG grid")
return channels
def get_channels_as_streaming_channel_list(self, nb_pages: int = 3) -> List:
"""
Get channels as StreamingChannel objects for the provider interface.
"""
from ...base.models import StreamingChannel
channels = self.get_channels(nb_pages=nb_pages)
streaming_channels = []
for ch in channels:
streaming_channel = StreamingChannel(
name=ch["title"],
content_id=ch["slug"],
provider=self._provider.provider_name,
logo_url=self._resolve_image_url(ch.get("logo_id")),
mode="live",
session_manifest=True,
manifest=None,
manifest_script=ch.get("seo"),
content_type="LIVE",
country="DE",
language="de",
)
streaming_channels.append(streaming_channel)
return streaming_channels
@staticmethod
def _resolve_image_url(image_id: str) -> str:
"""Resolve image ID to actual URL."""
if not image_id:
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.auth.get_bearer_token()
bedrock_token = self.auth.get_bedrock_token()
url = self.cfg.get_bedrock_layout_url(channel_seo=channel_seo)
location = f"https://plus.rtl.de/{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)
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":
video = item.get("itemContent", {}).get("video", {})
assets = video.get("assets", [])
if assets:
logger.debug(f"Found {len(assets)} assets for channel {channel_seo}")
return assets
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()
# --------------------------------------------------------------------------
# Manifest & DRM
# --------------------------------------------------------------------------
def get_best_manifest_url(
self,
channel_seo: str,
preferred_quality: str = None,
preferred_format: str = None,
preferred_drm_type: str = None,
) -> Optional[str]:
"""Get the best manifest URL for a channel based on preferences."""
assets = self.extract_stream_assets(channel_seo)
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")
# 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: {manifest_url}")
return manifest_url
# 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: {manifest_url}")
return manifest_url
# 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: {manifest_url}")
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: {manifest_url}")
return manifest_url
logger.error(f"No manifest URL found for channel {channel_seo}")
return None
def get_drm_config_for_channel(
self,
channel_seo: str,
preferred_quality: str = None,
) -> List[DRMConfig]:
"""
Get DRM configuration for a linear TV channel (Widevine + PlayReady).
Note: FairPlay (HLS) is currently not fully implemented as it requires
additional certificate handling. Only Widevine (DASH) is fully supported.
"""
assets = self.extract_stream_assets(channel_seo)
quality = preferred_quality or next(iter(self.cfg.preferred_qualities), "hd")
drm_configs = []
# Process DASH (Widevine) first - this is the primary DRM
for asset in assets:
if asset.get("format") == "dashcenc" and asset.get("quality") == quality:
drm_info = asset.get("drm", {})
drm_config = drm_info.get("config", {})
content_id = drm_config.get("contentId")
if not content_id:
logger.warning(f"No contentId in DRM config for {channel_seo}")
continue
try:
uid = self._get_user_id()
upfront_token = self.auth.get_upfront_token(content_id, uid)
license_url = self.cfg.drmtoday_license_url
headers = self.cfg.get_drm_license_headers(upfront_token)
drm_configs.append(DRMConfig(
system=DRMSystem.WIDEVINE,
priority=3,
license=LicenseConfig(
server_url=license_url,
req_headers=json.dumps(headers),
req_data="{CHA-RAW}",
use_http_get_request=False,
),
))
logger.debug(f"Added Widevine DRM for {channel_seo}")
except Exception as e:
logger.error(f"Failed to get Widevine DRM for {channel_seo}: {e}")
break # Only need one DASH asset
# Note: PlayReady/FairPlay (HLS) support can be added here if needed
# The infrastructure exists in constants.py (get_playready_drm_headers)
# but requires additional implementation and testing.
logger.info(f"Built {len(drm_configs)} DRM configs for {channel_seo}")
return drm_configs
# --------------------------------------------------------------------------
# Channel Info Helpers
# --------------------------------------------------------------------------
def get_channel_info(self, channel_seo: str) -> Optional[Dict[str, Any]]:
"""Get basic channel information from the layout."""
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_seo}: {e}")
return None
def get_current_program(self, channel_seo: str) -> Optional[Dict[str, Any]]:
"""Get currently playing program info for a channel."""
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", {})
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
except Exception as e:
logger.error(f"Failed to get current program for {channel_seo}: {e}")
return None
lib/streaming_providers/providers/rtlplus/constants.py
+190 -256
View File
@@ -7,18 +7,9 @@ from ..globals import get_user_agent
import json
import urllib.request
from typing import Optional
from ...base.utils.logger import logger
# Fetch RTL+ client version once at import time.
_RTLPLUS_CONFIG_URL = "https://plus.rtl.de/assets/config/config.json"
_RTLPLUS_CLIENT_VERSION_FALLBACK = "2025.6.26.0"
try:
with urllib.request.urlopen(_RTLPLUS_CONFIG_URL, timeout=4) as _resp:
_RTLPLUS_CLIENT_VERSION = json.loads(_resp.read().decode())["version"]
logger.debug(f"Fetched RTL+ client version: {_RTLPLUS_CLIENT_VERSION}")
except Exception as _exc:
logger.warning(f"Could not fetch RTL+ client version ({_exc}); using fallback {_RTLPLUS_CLIENT_VERSION_FALLBACK}")
_RTLPLUS_CLIENT_VERSION = _RTLPLUS_CLIENT_VERSION_FALLBACK
class RTLPlusDefaults:
"""Default values for RTL+ provider"""
@@ -30,24 +21,19 @@ class RTLPlusDefaults:
DEVICE_NAME = "Linux Chrome"
PLAYREADY_DEVICE_NAME = "Windows Edge"
# User agents — sourced from globals.get_user_agent() so versions stay current.
# Windows Chrome is used for standard requests; Windows Edge for PlayReady
# license acquisition (SOAP/XML), which requires an Edge UA.
# User agents
USER_AGENT = get_user_agent("windows", "chrome")
PLAYREADY_USER_AGENT = get_user_agent("windows", "edge")
# PlayReady SOAP action for license acquisition
# PlayReady SOAP action
PLAYREADY_SOAP_ACTION = "http://schemas.microsoft.com/DRM/2007/03/protocols/AcquireLicense"
# Supported platform identifiers
PLATFORM_WEB = "web"
PLATFORM_ANDROID = "android"
PLATFORM_IOS = "ios"
PLATFORM_SMART_TV = "smarttv"
PLATFORM_DEFAULT = PLATFORM_WEB
# Client and version information
CLIENT_VERSION = _RTLPLUS_CLIENT_VERSION
CLIENT_VERSION_FALLBACK = "2025.6.26.0"
CLIENT_ID = f"rtlplus-{PLATFORM_DEFAULT}"
# API endpoints
@@ -55,18 +41,21 @@ class RTLPlusDefaults:
AUTH_ENDPOINT = f"{AUTH_BASE_URL}/token"
AUTH_AUTHORIZE_ENDPOINT = f"{AUTH_BASE_URL}/auth"
GRAPHQL_ENDPOINT = "https://cdn.gateway.now-plus-prod.aws-cbc.cloud/graphql"
MANIFEST_ENDPOINT = (
"https://stus.player.streamingtech.de/livestream/linear/{channel_id}?platform={platform}"
)
EVENT_MANIFEST_ENDPOINT = (
"https://stus.player.streamingtech.de/liveevent/{content_id}?platform={platform}"
)
VOD_PLAYOUT_ENDPOINT = (
"https://stus.player.streamingtech.de/watch-playout-variants/{content_id}?platform={platform}"
)
BASE_WEBSITE = "https://plus.rtl.de/"
CONFIG_ENDPOINT = "https://plus.rtl.de/assets/config/config.json"
# Bedrock API endpoints (Linear TV)
BEDROCK_AUTH_URL = "https://front-auth.rtlde.bedrock.tech/v2/rtlde/platforms/m6group_web/token"
BEDROCK_LAYOUT_BASE = "https://layout.rtlde.bedrock.tech/front/v1/rtlde/m6group_web/main/token-web-31"
BEDROCK_DRM_UPFRONT_BASE = "https://drm.rtlde.bedrock.tech/v1/customers/rtlde/platforms/m6group_web/services/rtlplus_root/users/{uid}/live"
BEDROCK_HEARTBEAT_URL = "https://heartbeat-v2.rtlde.bedrock.tech/v2/platforms/m6group_web/notify/session_live"
# DRM license server
DRMTODAY_LICENSE_URL = "https://lic.drmtoday.com/license-proxy-widevine/cenc/"
# Direct CDN
ORIGIN_CDN_BASE = "https://origin.live.rtlde.bedrock.tech"
# Anonymous credentials (fallback)
ANONYMOUS_CLIENT_ID = "anonymous-user"
ANONYMOUS_CLIENT_SECRET = "4bfeb73f-1c4a-4e9f-a7fa-96aa1ad3d94c"
@@ -74,7 +63,7 @@ class RTLPlusDefaults:
# HTTP settings
DEFAULT_TIMEOUT = 30
# Custom header carrying the signed profile JWT (required by the events endpoint)
# Custom header carrying the signed profile JWT
PROFILE_HEADER = "Rtlplus-Profile"
# requiredPermission values returned by the events API
@@ -85,31 +74,27 @@ class RTLPlusDefaults:
TYPENAME_LIVE_EVENT_WIDGET = "LiveEventWidget"
TYPENAME_LIVE_EVENT = "LiveEvent"
# Linear TV Quality Presets
LINEAR_TV_PREFERRED_QUALITIES = ["hd", "sd"]
LINEAR_TV_PREFERRED_FORMATS = ["dashcenc", "hlsfp"]
LINEAR_TV_PREFERRED_DRM_TYPES = ["hardware", "software"]
# ---------------------------------------------------------------------------
# VOD — stream config endpoints
# ---------------------------------------------------------------------------
# Preferred stream-variant order for VOD (first match wins)
VOD_PREFERRED_VARIANTS = ["dashhd", "dashsd", "hlsfairplayhd", "hlsfairplaysd"]
# WatchPlayerConfigV3 returns stream URLs + DRM config directly from GraphQL
# (replaces the watch-playout-variants REST endpoint for VOD)
VOD_PLATFORM_GRAPHQL = PLATFORM_DEFAULT.upper()
# Wurstland — secondary stream resolver, returns DASH/HLS URLs
# GET /config/{rrn}/{platform}
VOD_WURSTLAND_CONFIG_URL = "https://wurstland.plus.rtl.de/config/{rrn}/{platform}"
VOD_WURSTLAND_PLATFORM = PLATFORM_DEFAULT.upper()
VOD_WURSTLAND_PLATFORM = PLATFORM_DEFAULT.upper()
# ---------------------------------------------------------------------------
# VOD — root category definition
# ---------------------------------------------------------------------------
VOD_ROOT_SLUG = "topic-worlds"
# Movies root / genre constants
VOD_MOVIES_ROOT_WATCH_PATH = "/video-tv/filme"
VOD_MOVIES_ROOT_WATCH_PATH = "/video-tv/filme"
VOD_MOVIES_GENRE_WATCH_PATH = "/video-tv/filme/genre"
VOD_MOVIES_ROOT_ID = "/video-tv/filme"
VOD_MOVIES_ROOT_ID = "/video-tv/filme"
VOD_MOVIE_GENRE_SLUGS = [
"action", "abenteuer", "animation", "comedy", "dokumentation",
@@ -117,10 +102,9 @@ class RTLPlusDefaults:
"liebesfilm", "science-fiction", "thriller",
]
# Series root / genre constants
VOD_SERIES_ROOT_WATCH_PATH = "/video-tv/serien"
VOD_SERIES_ROOT_WATCH_PATH = "/video-tv/serien"
VOD_SERIES_GENRE_WATCH_PATH = "/video-tv/serien/genre"
VOD_SERIES_ROOT_ID = "/video-tv/serien"
VOD_SERIES_ROOT_ID = "/video-tv/serien"
VOD_SERIES_GENRE_SLUGS = [
"action", "animation", "comedy", "crime", "dokumentation",
@@ -128,143 +112,107 @@ class RTLPlusDefaults:
"romance", "science-fiction", "thriller",
]
# Shows root / genre constants
VOD_SHOWS_ROOT_WATCH_PATH = "/video-tv/shows"
VOD_SHOWS_ROOT_WATCH_PATH = "/video-tv/shows"
VOD_SHOWS_GENRE_WATCH_PATH = "/video-tv/shows/genre"
VOD_SHOWS_ROOT_ID = "/video-tv/shows"
VOD_SHOWS_ROOT_ID = "/video-tv/shows"
VOD_SHOW_GENRE_SLUGS = [
"action", "comedy", "crime", "dokumentation",
"drama", "kinder", "reality", "romance", "thriller",
]
# OverviewPage elementType values and pagination
VOD_ELEMENT_TYPE_MOVIE = "MOVIE"
VOD_ELEMENT_TYPE_MOVIE = "MOVIE"
VOD_ELEMENT_TYPE_SERIES = "SERIES"
VOD_ELEMENT_TYPE_SHOW = "SHOW"
VOD_ELEMENT_TYPE_SHOW = "SHOW"
VOD_OVERVIEW_PAGE_LIMIT = 48
# Lazy-loaded client version
_CLIENT_VERSION_CACHE: Optional[str] = None
def _get_rtlplus_client_version() -> str:
"""Lazy fetch RTL+ client version from config endpoint."""
global _CLIENT_VERSION_CACHE
if _CLIENT_VERSION_CACHE is None:
try:
with urllib.request.urlopen(RTLPlusDefaults.CONFIG_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}")
except Exception as exc:
logger.warning(f"Could not fetch RTL+ client version ({exc}); using fallback {RTLPlusDefaults.CLIENT_VERSION_FALLBACK}")
_CLIENT_VERSION_CACHE = RTLPlusDefaults.CLIENT_VERSION_FALLBACK
return _CLIENT_VERSION_CACHE
class RTLPlusGraphQL:
"""
Persisted-query hashes and parameter builders for all RTL+ GraphQL operations.
"""Persisted-query hashes and parameter builders for all RTL+ GraphQL operations."""
The HASHES dict maps each GraphQL operationName to its sha256 hash as
captured from live browser network traffic. Every builder method calls
_build(), which assembles the standard persisted-query envelope so that
the boilerplate never has to be repeated.
"""
# sha256 hashes keyed by operationName (captured from browser network traffic)
HASHES: dict[str, str] = {
"LiveTvStations": "845cf56a2a78110a0f978c1a2af2bc7f9a1c937d0f324ffaf852a9a4414c8485",
"LiveTvStations": "845cf56a2a78110a0f978c1a2af2bc7f9a1c937d0f324ffaf852a9a4414c8485",
"ExploreWidgetWatch": "724f21ab86aa3f8c57673a3b346cf119f6a155bf82bb0201fd3b18e28e44f1ed",
"TopicWorlds": "3dbde4c45532f4bdb0a1d7c210db43f0888f8a82f4aab43f7a90f3c4762d8ff7",
"Format": "d112638c0184ab5698af7b69532dfe2f12973f7af9cb137b9f70278130b1eafa",
"MRE": "0c77404637570adff548e329a48654498c54ce1c36a459d72586ff18999bebaa",
"Episode": "87dbde15a0d269b11606f5ff458d555e98eb493bb4fb6ddc150d812d5e9a9cf8",
"WatchPlayerConfigV3":"fea0311fb572b6fded60c5a1a9d652f97f55d182bc4cedbdad676354a8d2797c",
"SeoUrlData": "fcc4a812d6b93496f00c3068234db7722f553032bb760e09e5e6c74586c86f8d",
"OverviewPage": "28aad4e992bb63330bfcd40a6906af3119d8a2612fa9fd28dae9c19127e247ca",
"LiveEventsOverviewPage": "77a8f26d5de76daf801fcd7ae54bebd0ecabbeeb3ecb90889bc4a2e5590b7d20",
"SeasonWithFormatAndEpisodes": "cc0fbbe17143f549a35efa6f8665ceb9b1cfae44b590f0b2381a9a304304c584",
"TopicWorlds": "3dbde4c45532f4bdb0a1d7c210db43f0888f8a82f4aab43f7a90f3c4762d8ff7",
"Format": "d112638c0184ab5698af7b69532dfe2f12973f7af9cb137b9f70278130b1eafa",
"MRE": "0c77404637570adff548e329a48654498c54ce1c36a459d72586ff18999bebaa",
"Episode": "87dbde15a0d269b11606f5ff458d555e98eb493bb4fb6ddc150d812d5e9a9cf8",
"WatchPlayerConfigV3": "fea0311fb572b6fded60c5a1a9d652f97f55d182bc4cedbdad676354a8d2797c",
"SeoUrlData": "fcc4a812d6b93496f00c3068234db7722f553032bb760e09e5e6c74586c86f8d",
"OverviewPage": "28aad4e992bb63330bfcd40a6906af3119d8a2612fa9fd28dae9c19127e247ca",
"LiveEventsOverviewPage": "77a8f26d5de76daf801fcd7ae54bebd0ecabbeeb3ecb90889bc4a2e5590b7d20",
"SeasonWithFormatAndEpisodes": "cc0fbbe17143f549a35efa6f8665ceb9b1cfae44b590f0b2381a9a304304c584",
}
@classmethod
def _build(cls, operation: str, variables: str) -> dict:
"""Assemble a persisted-query parameter dict for *operation*."""
return {
"operationName": operation,
"variables": variables,
"extensions": (
f'{{"persistedQuery":{{"version":1,"sha256Hash":"{cls.HASHES[operation]}"}}}}'
),
"extensions": f'{{"persistedQuery":{{"version":1,"sha256Hash":"{cls.HASHES[operation]}"}}}}',
}
# --- Live TV / Events ---
@classmethod
def live_tv_stations(cls, epg_count: int = 4) -> dict:
"""LiveTvStations — all broadcast + FAST channels with EPG slots."""
return cls._build(
"LiveTvStations",
f'{{"epgCount":{epg_count},"filter":{{"channelTypes":["BROADCAST","FAST"]}}}}',
)
return cls._build("LiveTvStations", f'{{"epgCount":{epg_count},"filter":{{"channelTypes":["BROADCAST","FAST"]}}}}')
@classmethod
def explore_widget_watch(cls, area: str = "home", offset: int = 0, take: int = 15) -> dict:
"""ExploreWidgetWatch — editorial home view containing LiveEventWidget nodes."""
return cls._build(
"ExploreWidgetWatch",
f'{{"area":"{area}","offset":{offset},"take":{take}}}',
)
return cls._build("ExploreWidgetWatch", f'{{"area":"{area}","offset":{offset},"take":{take}}}')
@classmethod
def season_with_episodes(cls, season_rrn: str, offset: int = 0, limit: int = 48) -> dict:
"""SeasonWithFormatAndEpisodes — episodes for a specific season RRN."""
return cls._build(
"SeasonWithFormatAndEpisodes",
f'{{"seasonId":"{season_rrn}","offset":{offset},"limit":{limit}}}',
)
return cls._build("SeasonWithFormatAndEpisodes", f'{{"seasonId":"{season_rrn}","offset":{offset},"limit":{limit}}}')
@classmethod
def live_events_overview_page(cls) -> dict:
"""LiveEventsOverview — all upcoming/live events grouped by teaserRows."""
return cls._build("LiveEventsOverviewPage", "{}")
# --- VOD ---
@classmethod
def topic_worlds(cls, take: int = 100, offset: int = 0) -> dict:
"""TopicWorlds — browse all genre/topic worlds.
Confirmed from live traffic:
response key: data.topicWorldsV2.elements[] (NOT topicWorlds.items[])
pagination: data.topicWorldsV2.pageInfo.hasNextPage
"""
return cls._build(
"TopicWorlds",
f'{{"take":{take},"offset":{offset},"filterForSearchGrid":true}}',
)
return cls._build("TopicWorlds", f'{{"take":{take},"offset":{offset},"filterForSearchGrid":true}}')
@classmethod
def format(cls, format_rrn: str) -> dict:
"""Format — full format detail (seasons, metadata) by RRN."""
return cls._build("Format", f'{{"id":"{format_rrn}"}}')
@classmethod
def mre(cls, format_rrn: str) -> dict:
"""MRE — seasons + related episodes for a Format."""
return cls._build("MRE", f'{{"id":"{format_rrn}"}}')
@classmethod
def episode(cls, episode_rrn: str) -> dict:
"""Episode — full episode detail by RRN."""
return cls._build("Episode", f'{{"rrn":"{episode_rrn}"}}')
@classmethod
def watch_player_config(cls, rrn: str) -> dict:
"""WatchPlayerConfigV3 — stream URL + DRM config for an episode or movie."""
return cls._build(
"WatchPlayerConfigV3",
f'{{"platform":"{RTLPlusDefaults.VOD_PLATFORM_GRAPHQL}","id":"{rrn}"}}',
)
return cls._build("WatchPlayerConfigV3", f'{{"platform":"{RTLPlusDefaults.VOD_PLATFORM_GRAPHQL}","id":"{rrn}"}}')
@classmethod
def seo_url_data(cls, watch_path: str) -> dict:
"""SeoUrlData — resolve a watch-path to its content hierarchy."""
return cls._build("SeoUrlData", f'{{"watchPath":"{watch_path}"}}')
@classmethod
def overview_page(
cls,
element_type: str,
genres: list = None,
offset: int = 0,
limit: int = None,
) -> dict:
"""OverviewPage — paginated grid for MOVIE or SERIES with optional genre filter."""
def overview_page(cls, element_type: str, genres: list = None, offset: int = 0, limit: int = None) -> dict:
if limit is None:
limit = RTLPlusDefaults.VOD_OVERVIEW_PAGE_LIMIT
@@ -273,10 +221,7 @@ class RTLPlusGraphQL:
genres_json = "[" + ",".join(f'"{g}"' for g in genres) + "]"
filter_parts.append(f'"genres":{genres_json}')
variables = (
f'{{"pagination":{{"offset":{offset},"limit":{limit}}},'
f'"filter":{{{",".join(filter_parts)}}}}}'
)
variables = f'{{"pagination":{{"offset":{offset},"limit":{limit}}},"filter":{{{",".join(filter_parts)}}}}}'
return cls._build("OverviewPage", variables)
@@ -285,95 +230,96 @@ class RTLPlusHeaders:
@staticmethod
def get_base_headers(user_agent: str = None) -> dict:
"""Get base HTTP headers"""
return {
"User-Agent": user_agent or RTLPlusDefaults.USER_AGENT,
"Accept": "application/json",
}
return {"User-Agent": user_agent or RTLPlusDefaults.USER_AGENT, "Accept": "application/json"}
@staticmethod
def get_auth_headers(user_agent: str = None) -> dict:
"""Get headers for authentication requests"""
headers = RTLPlusHeaders.get_base_headers(user_agent)
headers.update(
{
"Content-Type": "application/x-www-form-urlencoded",
"Origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"Referer": RTLPlusDefaults.BASE_WEBSITE,
}
)
headers.update({
"Content-Type": "application/x-www-form-urlencoded",
"Origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"Referer": RTLPlusDefaults.BASE_WEBSITE,
})
return headers
@staticmethod
def get_api_headers(
access_token: str = None,
device_id: str = None,
client_version: str = None,
user_agent: str = None,
) -> dict:
"""Get headers for authenticated API requests"""
def get_api_headers(access_token: str = None, device_id: str = None, client_version: str = None, user_agent: str = None) -> dict:
headers = RTLPlusHeaders.get_base_headers(user_agent)
headers.update(
{
"Content-Type": "application/json",
"Rtlplus-Client-Id": f"rci:rtlplus:{RTLPlusDefaults.PLATFORM_DEFAULT}",
"Rtlplus-Referrer": "",
"Rtlplus-Client-Version": client_version or RTLPlusDefaults.CLIENT_VERSION,
}
)
headers.update({
"Content-Type": "application/json",
"Rtlplus-Client-Id": f"rci:rtlplus:{RTLPlusDefaults.PLATFORM_DEFAULT}",
"Rtlplus-Referrer": "",
"Rtlplus-Client-Version": client_version or _get_rtlplus_client_version(),
})
if access_token:
headers["Authorization"] = f"Bearer {access_token}"
if device_id:
headers["X-Device-Id"] = device_id
return headers
@staticmethod
def get_events_headers(
access_token: str,
profile_token: str = None,
device_id: str = None,
client_version: str = None,
user_agent: str = None,
) -> dict:
"""Get headers for the editorial GraphQL endpoint (events / ExploreWidgetWatch).
Extends the standard API headers with the optional ``Rtlplus-Profile``
JWT that the browser sends when a user profile is active. The header
is omitted when *profile_token* is ``None`` so that anonymous / client-
credential sessions work without modification.
Args:
access_token: Bearer token from the auth endpoint.
profile_token: Signed profile JWT (``Rtlplus-Profile`` header).
Pass ``None`` for anonymous / client-credential sessions.
device_id: Device UUID forwarded as ``X-Device-Id``.
client_version: Client version string; falls back to the default.
user_agent: UA string; falls back to the default.
Returns:
dict of HTTP headers ready for use with the GraphQL endpoint.
"""
headers = RTLPlusHeaders.get_api_headers(
access_token=access_token,
device_id=device_id,
client_version=client_version,
user_agent=user_agent,
)
def get_events_headers(access_token: str, profile_token: str = None, device_id: str = None, client_version: str = None, user_agent: str = None) -> dict:
headers = RTLPlusHeaders.get_api_headers(access_token=access_token, device_id=device_id, client_version=client_version, user_agent=user_agent)
if profile_token:
headers[RTLPlusDefaults.PROFILE_HEADER] = profile_token
return headers
@staticmethod
def get_bedrock_layout_headers(
oauth_token: str,
bedrock_token: str,
device_id: str,
client_version: str,
user_agent: str,
location: str = None,
) -> dict:
headers = {
"authorization": f"Bearer {oauth_token}",
"cache-control": "no-cache",
"origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"referer": RTLPlusDefaults.BASE_WEBSITE,
"request-timeout": "10000",
"user-agent": user_agent,
"x-bedrock-token": bedrock_token,
"x-auth-device-id": device_id,
"x-auth-device-name": RTLPlusDefaults.DEVICE_NAME,
"x-client-release": client_version,
"x-customer-name": "rtlde",
}
if location:
headers["x-location"] = location
return headers
@staticmethod
def get_upfront_token_headers(oauth_token: str, bedrock_token: str, device_id: str, client_version: str, user_agent: str) -> dict:
return {
"authorization": f"Bearer {oauth_token}",
"cache-control": "no-cache",
"origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"referer": RTLPlusDefaults.BASE_WEBSITE,
"request-timeout": "10000",
"user-agent": user_agent,
"x-bedrock-token": bedrock_token,
"x-auth-device-id": device_id,
"x-auth-device-name": RTLPlusDefaults.DEVICE_NAME,
"x-client-release": client_version,
"x-customer-name": "rtlde",
}
@staticmethod
def get_drm_license_headers(upfront_token: str, user_agent: str) -> dict:
return {
"accept": "*/*",
"content-type": "application/json",
"origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"referer": RTLPlusDefaults.BASE_WEBSITE,
"user-agent": user_agent,
"x-dt-auth-token": upfront_token,
}
@staticmethod
def get_drm_headers(access_token: str, device_id: str = None, user_agent: str = None) -> dict:
return {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br, zstd",
"Accept-Language": "de-DE,de;q=0.9",
"Cache-Control": "no-cache",
"Origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"Referer": RTLPlusDefaults.BASE_WEBSITE,
@@ -383,21 +329,8 @@ class RTLPlusHeaders:
"X-Device-Name": RTLPlusDefaults.DEVICE_NAME,
}
@staticmethod
def get_event_manifest_headers(access_token: str, user_agent: str = None) -> dict:
"""Get headers for the event manifest endpoint (stus.player.streamingtech.de/liveevent).
Uses X-Auth-Token scheme as required by this endpoint."""
return {
"X-Auth-Token": access_token,
"Content-Type": "application/json",
"Origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
"Referer": RTLPlusDefaults.BASE_WEBSITE,
"User-Agent": user_agent or RTLPlusDefaults.USER_AGENT,
}
@staticmethod
def get_playready_drm_headers(access_token: str, device_id: str = None) -> dict:
"""Get headers for PlayReady DRM license acquisition (SOAP/XML)"""
return {
"Content-Type": "text/xml; charset=UTF-8",
"Origin": RTLPlusDefaults.BASE_WEBSITE.rstrip("/"),
@@ -414,52 +347,58 @@ class RTLPlusConfig:
"""Configuration class that can be customized per instance"""
def __init__(self, config_dict: dict = None):
"""Initialize with optional configuration overrides"""
config = config_dict or {}
self.logo = config.get("logo", RTLPlusDefaults.RTLPLUS_LOGO)
# Core settings (can be overridden)
self.client_version = config.get("client_version", RTLPlusDefaults.CLIENT_VERSION)
# Core settings
self.client_version = config.get("client_version", _get_rtlplus_client_version())
self.device_id = config.get("device_id", RTLPlusDefaults.DEVICE_ID)
self.user_agent = config.get("user_agent", RTLPlusDefaults.USER_AGENT)
self.platform = config.get("platform", RTLPlusDefaults.PLATFORM_DEFAULT)
# API endpoints (can be overridden for testing)
# API endpoints
self.auth_endpoint = config.get("auth_endpoint", RTLPlusDefaults.AUTH_ENDPOINT)
self.graphql_endpoint = config.get("graphql_endpoint", RTLPlusDefaults.GRAPHQL_ENDPOINT)
self.manifest_endpoint = config.get("manifest_endpoint", RTLPlusDefaults.MANIFEST_ENDPOINT)
self.event_manifest_endpoint = config.get("event_manifest_endpoint", RTLPlusDefaults.EVENT_MANIFEST_ENDPOINT)
self.vod_playout_endpoint = config.get("vod_playout_endpoint", RTLPlusDefaults.VOD_PLAYOUT_ENDPOINT)
self.base_website = config.get("base_website", RTLPlusDefaults.BASE_WEBSITE)
self.config_endpoint = config.get("config_endpoint", RTLPlusDefaults.CONFIG_ENDPOINT)
# Bedrock endpoints (Linear TV)
self.bedrock_auth_url = config.get("bedrock_auth_url", RTLPlusDefaults.BEDROCK_AUTH_URL)
self.bedrock_layout_base = config.get("bedrock_layout_base", RTLPlusDefaults.BEDROCK_LAYOUT_BASE)
self.bedrock_drm_upfront_base = config.get("bedrock_drm_upfront_base", RTLPlusDefaults.BEDROCK_DRM_UPFRONT_BASE)
self.bedrock_heartbeat_url = config.get("bedrock_heartbeat_url", RTLPlusDefaults.BEDROCK_HEARTBEAT_URL)
# DRM license server
self.drmtoday_license_url = config.get("drmtoday_license_url", RTLPlusDefaults.DRMTODAY_LICENSE_URL)
# Quality presets for linear TV
self.preferred_qualities = config.get("preferred_qualities", RTLPlusDefaults.LINEAR_TV_PREFERRED_QUALITIES)
self.preferred_formats = config.get("preferred_formats", RTLPlusDefaults.LINEAR_TV_PREFERRED_FORMATS)
self.preferred_drm_types = config.get("preferred_drm_types", RTLPlusDefaults.LINEAR_TV_PREFERRED_DRM_TYPES)
# HTTP settings
self.timeout = config.get("timeout", RTLPlusDefaults.DEFAULT_TIMEOUT)
def get_manifest_url(self, content_id: str) -> str:
if "live-events" in content_id:
return self.event_manifest_endpoint.format(
content_id=content_id, platform=self.platform
)
if content_id.startswith("rrn:watch:videohub:station:"):
return self.manifest_endpoint.format(
channel_id=content_id, platform=self.platform
)
return self.vod_playout_endpoint.format(
content_id=content_id, platform=self.platform
)
def get_bedrock_layout_url(self, channel_seo: str) -> str:
"""Get Bedrock layout URL for a specific channel"""
return f"{self.bedrock_layout_base}/live/{channel_seo}/layout"
def get_epg_grid_url(self) -> str:
"""Get EPG grid URL for channel listing"""
return f"{self.bedrock_layout_base}/epg_grid"
def get_upfront_token_url(self, uid: str, content_id: str) -> str:
"""Get upfront token URL for DRM"""
return f"{self.bedrock_drm_upfront_base.format(uid=uid)}/{content_id}/upfront-token"
def get_base_headers(self) -> dict:
"""Get base headers with this config's user agent"""
return RTLPlusHeaders.get_base_headers(self.user_agent)
def get_auth_headers(self) -> dict:
"""Get auth headers with this config's settings"""
return RTLPlusHeaders.get_auth_headers(self.user_agent)
def get_api_headers(self, access_token: str = None) -> dict:
"""Get API headers with this config's settings"""
return RTLPlusHeaders.get_api_headers(
access_token=access_token,
device_id=self.device_id,
@@ -468,16 +407,6 @@ class RTLPlusConfig:
)
def get_events_headers(self, access_token: str, profile_token: str = None) -> dict:
"""
Get headers for the editorial GraphQL endpoint with this config's settings.
Args:
access_token: Bearer token from the auth endpoint.
profile_token: Optional signed profile JWT (``Rtlplus-Profile``).
Returns:
dict of HTTP headers.
"""
return RTLPlusHeaders.get_events_headers(
access_token=access_token,
profile_token=profile_token,
@@ -486,35 +415,40 @@ class RTLPlusConfig:
user_agent=self.user_agent,
)
def get_bedrock_layout_headers(self, oauth_token: str, bedrock_token: str, location: str = None) -> dict:
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,
)
def get_upfront_token_headers(self, oauth_token: str, bedrock_token: str) -> dict:
return RTLPlusHeaders.get_upfront_token_headers(
oauth_token=oauth_token,
bedrock_token=bedrock_token,
device_id=self.device_id,
client_version=self.client_version,
user_agent=self.user_agent,
)
def get_drm_license_headers(self, upfront_token: str) -> dict:
return RTLPlusHeaders.get_drm_license_headers(
upfront_token=upfront_token,
user_agent=self.user_agent,
)
def get_drm_headers(self, access_token: str) -> dict:
"""Get DRM headers with this config's settings"""
return RTLPlusHeaders.get_drm_headers(
access_token=access_token,
device_id=self.device_id,
user_agent=self.user_agent,
)
def get_event_manifest_headers(self, access_token: str) -> dict:
"""Get event manifest headers with this config's settings."""
return RTLPlusHeaders.get_event_manifest_headers(
access_token=access_token,
user_agent=self.user_agent,
)
def get_playready_drm_headers(self, access_token: str) -> dict:
"""Get PlayReady-specific DRM headers (Edge UA, SOAP content-type)"""
return RTLPlusHeaders.get_playready_drm_headers(
access_token=access_token,
device_id=self.device_id,
)
# ---------------------------------------------------------------------------
# VOD helpers
# ---------------------------------------------------------------------------
@staticmethod
def get_vod_wurstland_url(rrn: str) -> str:
"""Return the Wurstland config URL for a VOD RRN."""
return RTLPlusDefaults.VOD_WURSTLAND_CONFIG_URL.format(
rrn=rrn, platform=RTLPlusDefaults.VOD_WURSTLAND_PLATFORM
)
lib/streaming_providers/providers/rtlplus/models.py
+39 -28
View File
@@ -29,7 +29,6 @@ class RTLPlusUserCredentials(UserPasswordCredentials):
"username": self.username,
"password": self.password,
}
# RTL+ might need client_id for user auth - adjust based on API requirements
if self.client_id:
payload["client_id"] = self.client_id
return payload
@@ -65,7 +64,6 @@ class RTLPlusAuthToken(BaseAuthToken):
not_before_policy: Optional[int] = None,
scope: str = "",
):
# Initialize parent class with refresh_expires_in
super().__init__(
access_token=access_token,
token_type=token_type,
@@ -74,7 +72,6 @@ class RTLPlusAuthToken(BaseAuthToken):
refresh_token=refresh_token,
refresh_expires_in=refresh_expires_in,
)
# RTL+ specific fields
self.refresh_expires_in = refresh_expires_in
self.not_before_policy = not_before_policy
self.scope = scope
@@ -114,21 +111,36 @@ class RTLPlusAuthToken(BaseAuthToken):
return False
current_time = time.time()
# Add small buffer (30 seconds) to account for network delays
buffer_time = 30
return current_time < (self.issued_at + self.expires_in - buffer_time)
# Add this method to the RTLPlusAuthToken class in models.py
def is_anonymous_token(self) -> bool:
"""Check if this token was obtained via anonymous authentication"""
"""Check if this token was obtained via anonymous authentication."""
if not self.access_token:
return True
try:
# Simple check without full JWT decoding - look for anonymous client ID pattern
return "anonymous-user" in self.access_token
except:
import base64
import json
parts = self.access_token.split(".")
if len(parts) < 2:
return False
payload_segment = parts[1]
padding = 4 - len(payload_segment) % 4
if padding != 4:
payload_segment += "=" * padding
payload_json = base64.b64decode(payload_segment)
payload = json.loads(payload_json)
client_id = payload.get("clientId")
is_guest = payload.get("isGuest", False)
return is_guest and client_id == "anonymous-user"
except Exception:
return False
@@ -148,13 +160,11 @@ class RTLPlusChannel:
sort_order: int = 0
def __post_init__(self):
"""Validate channel data after initialization"""
if not self.id or not self.name:
raise ValueError("Channel must have both id and name")
@classmethod
def from_api_response(cls, data: Dict[str, Any]) -> "RTLPlusChannel":
"""Create channel from API response data"""
return cls(
id=data["id"],
name=data["name"],
@@ -167,7 +177,6 @@ class RTLPlusChannel:
)
def to_dict(self) -> Dict[str, Any]:
"""Convert channel to dictionary"""
return {
"id": self.id,
"name": self.name,
@@ -179,16 +188,17 @@ class RTLPlusChannel:
"sort_order": self.sort_order,
}
@dataclass
class RTLPlusLiveEvent:
"""
RTL+ API representation of a live event before conversion to the
domain Event model. Keeps parsing logic isolated from the domain layer.
domain Event model.
"""
id: str
title: str
stream_start: str # ISO string from API
stream_end: str # ISO string from API
stream_start: str
stream_end: str
location: Optional[str]
event_category: str
event_sub_category: str
@@ -200,9 +210,8 @@ class RTLPlusLiveEvent:
@classmethod
def from_api_node(cls, node: Dict[str, Any]) -> "RTLPlusLiveEvent":
"""Parse a single LiveEvent GraphQL node from LiveEventsOverview."""
details = node.get("details") or {}
images = node.get("images") or {}
images = node.get("images") or {}
return cls(
id=node["id"],
title=node["title"],
@@ -217,18 +226,23 @@ class RTLPlusLiveEvent:
images.get("artworkLandscape", {}).get("url")
or images.get("artworkPortrait", {}).get("url")
),
required_permission=node.get(
"requiredPermission", RTLPlusDefaults.PERMISSION_PAY_TV
),
required_permission=node.get("requiredPermission", RTLPlusDefaults.PERMISSION_PAY_TV),
watch_path=node.get("urlData", {}).get("watchPath", ""),
)
def to_event(self, provider: str = "rtlplus") -> "Event":
"""Convert to the domain Event model."""
from dateutil.parser import isoparse # already likely a dep; fallback below
"""Convert to the domain Event model with lazy dateutil import."""
try:
from dateutil.parser import isoparse
except ImportError:
# Fallback for environments without dateutil
from datetime import datetime
def isoparse(date_string):
# Simple ISO parser for common formats
return datetime.fromisoformat(date_string.replace('Z', '+00:00'))
start = isoparse(self.stream_start)
end = isoparse(self.stream_end)
end = isoparse(self.stream_end)
return Event(
name=self.title,
@@ -239,12 +253,13 @@ class RTLPlusLiveEvent:
logo_url=self.logo_url,
description=self.description,
genre=self.event_sub_category,
content_type=ContentType.LIVE, # always live for RTL+ events
content_type=ContentType.LIVE,
language="de",
country="DE",
venue=self.location,
)
@dataclass
class RTLPlusStreamInfo:
"""
@@ -259,13 +274,11 @@ class RTLPlusStreamInfo:
quality: str = "auto"
def __post_init__(self):
"""Validate stream info after initialization"""
if not self.manifest_url or not self.channel_id:
raise ValueError("Stream must have both manifest_url and channel_id")
@classmethod
def from_manifest_response(cls, data: Dict[str, Any], channel_id: str) -> "RTLPlusStreamInfo":
"""Create stream info from manifest API response"""
return cls(
manifest_url=data["url"],
channel_id=channel_id,
@@ -276,7 +289,6 @@ class RTLPlusStreamInfo:
)
def to_dict(self) -> Dict[str, Any]:
"""Convert stream info to dictionary"""
return {
"manifest_url": self.manifest_url,
"channel_id": self.channel_id,
@@ -287,5 +299,4 @@ class RTLPlusStreamInfo:
}
def has_drm(self) -> bool:
"""Check if stream has DRM protection"""
return bool(self.drm_license_url and self.drm_key_id)
lib/streaming_providers/providers/rtlplus/provider.py
+115 -272
View File
@@ -1,7 +1,7 @@
# lib/streaming_providers/providers/rtlplus/provider.py
# streaming_providers/providers/rtlplus/provider.py
import json
import time
import datetime
from datetime import datetime
from typing import ClassVar, Dict, List, Optional, Any
import requests
@@ -14,8 +14,9 @@ from .auth import RTLPlusAuthenticator
from .constants import RTLPlusConfig, RTLPlusDefaults, RTLPlusGraphQL
from .models import RTLPlusLiveEvent
from .vod_manager import RTLPlusVodManager
from .channel_manager import RTLPlusChannelManager
_MANIFEST_CACHE_TTL = 86400 # 1 day in seconds
_MANIFEST_CACHE_TTL = 86400 # 1 day in seconds for VOD/events
class RTLPlusProvider(StreamingProvider):
@@ -28,18 +29,17 @@ class RTLPlusProvider(StreamingProvider):
]
def __init__(
self,
country: str = "DE",
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
self,
country: str = "DE",
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
):
super().__init__(country)
# Initialize configuration
self.rtl_config = RTLPlusConfig(config)
self.channels_query_params = RTLPlusGraphQL.live_tv_stations()
# ✅ Using HTTP manager abstraction
# Setup HTTP manager
self.http_manager = self._setup_http_manager(
provider_name="rtlplus",
proxy_config=proxy_config,
@@ -55,20 +55,21 @@ class RTLPlusProvider(StreamingProvider):
http_manager=self.http_manager,
)
# ✅ Share HTTP manager with authenticator
self.http_manager = self._share_http_manager_with_authenticator(self.authenticator)
# Try authentication
try:
self.bearer_token = self.authenticator.get_bearer_token()
logger.debug(f"RTL+ authentication successful during initialization")
logger.debug("RTL+ authentication successful during initialization")
except Exception as e:
logger.warning(f"RTL+ could not authenticate during initialization: {e}")
self.bearer_token = None
# Initialize managers
self._vod_manager = RTLPlusVodManager(self)
self.channel_manager = RTLPlusChannelManager(self)
# Manifest cache: content_id -> (data, timestamp)
# Manifest cache for VOD/events
self._manifest_cache: Dict[str, tuple] = {}
@property
@@ -85,7 +86,6 @@ class RTLPlusProvider(StreamingProvider):
@property
def uses_dynamic_manifests(self) -> bool:
# RTL+ provides relatively stable manifest URLs that can be fetched and cached
return False
@property
@@ -96,103 +96,53 @@ class RTLPlusProvider(StreamingProvider):
def supported_auth_types(self) -> List[str]:
return ["user_credentials"]
# ============================================================================
# OPTION 1: Provider-specific method (RECOMMENDED - No signature conflict)
# ============================================================================
def _get_rtlplus_authenticated_headers(self) -> Dict[str, str]:
"""
Get headers with authentication and RTL+ specific headers.
# --------------------------------------------------------------------------
# Authentication Helpers
# --------------------------------------------------------------------------
Upgrades from anonymous to user token only when user credentials are
configured and the current token is not already at user level. This
prevents a full re-authentication round-trip on every single request
once a valid user token is already held.
"""
def _get_rtlplus_authenticated_headers(self) -> Dict[str, str]:
from ...base.auth.base_auth import TokenAuthLevel
current_level = self.authenticator.get_current_token_level()
force_upgrade = (
self.authenticator.has_user_credentials()
and current_level != TokenAuthLevel.USER_AUTHENTICATED
)
bearer_token = self.authenticator.get_bearer_token(force_upgrade=force_upgrade)
try:
current_level = self.authenticator.get_current_token_level()
force_upgrade = (
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:
logger.error(f"Failed to get bearer token: {e}")
raise
return self.rtl_config.get_api_headers(access_token=bearer_token)
def _get_event_manifest_headers(self) -> Dict[str, str]:
access_token = self.authenticator.get_bearer_token()
return self.rtl_config.get_event_manifest_headers(access_token=access_token)
# --------------------------------------------------------------------------
# Linear TV Channels
# --------------------------------------------------------------------------
def get_channels(self, **kwargs) -> List[StreamingChannel]:
"""
Fetch channels from RTL+ GraphQL API with authentication
Get list of available linear TV channels from EPG grid.
"""
try:
# ✅ Use provider-specific method
headers = self._get_rtlplus_authenticated_headers()
response = self.http_manager.get(
self.rtl_config.graphql_endpoint,
operation="api",
params=self.channels_query_params,
headers=headers,
)
response.raise_for_status()
data = response.json()
channels = []
if "data" in data and "liveTvStations" in data["data"]:
for station in data["data"]["liveTvStations"]:
channel = self._parse_station_to_channel(station)
if channel:
channels.append(channel)
channels = self.channel_manager.get_channels_as_streaming_channel_list()
self.channels = channels
return channels
except requests.RequestException as e:
logger.error(f"Error fetching RTL+ channels: {e}")
# Try to refresh auth token and retry once
try:
logger.info("Attempting to refresh authentication and retry...")
self.authenticator.invalidate_token()
headers = self._get_rtlplus_authenticated_headers()
response = self.http_manager.get(
self.rtl_config.graphql_endpoint,
operation="api",
params=self.channels_query_params,
headers=headers,
)
response.raise_for_status()
data = response.json()
channels = []
if "data" in data and "liveTvStations" in data["data"]:
for station in data["data"]["liveTvStations"]:
channel = self._parse_station_to_channel(station)
if channel:
channels.append(channel)
self.channels = channels
return channels
except Exception as retry_e:
logger.error(f"Retry failed: {retry_e}")
return []
except Exception as e:
logger.error(f"Error parsing RTL+ channels: {e}")
logger.error(f"Error fetching RTL+ channels: {e}")
return []
# --------------------------------------------------------------------------
# Events (Live Events)
# --------------------------------------------------------------------------
def get_events(
self,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**kwargs,
) -> List[Event]:
"""
Fetch upcoming / live RTL+ events from the editorial GraphQL endpoint.
Filters by start_time / end_time if provided (both timezone-aware).
"""
"""Fetch upcoming / live RTL+ events from the editorial GraphQL endpoint."""
raw_events = self._fetch_live_events()
events: List[Event] = []
@@ -203,7 +153,6 @@ class RTLPlusProvider(StreamingProvider):
logger.warning(f"RTL+: Could not convert event '{raw.id}': {e}")
continue
# Optional time-window filtering (delegate to caller semantics)
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:
@@ -233,20 +182,10 @@ class RTLPlusProvider(StreamingProvider):
@staticmethod
def _parse_live_events(data: Dict[str, Any]) -> List[RTLPlusLiveEvent]:
"""
Extract LiveEvent nodes from the LiveEventsOverview GraphQL response.
Response shape: data.liveEventsOverview.teaserRows[].events[]
Static so it's unit-testable without provider state.
"""
events: List[RTLPlusLiveEvent] = []
seen: set = set()
teaser_rows = (
data.get("data", {})
.get("liveEventsOverview", {})
.get("teaserRows", [])
)
teaser_rows = data.get("data", {}).get("liveEventsOverview", {}).get("teaserRows", [])
for row in teaser_rows:
for element in (row.get("events") or []):
@@ -265,102 +204,42 @@ class RTLPlusProvider(StreamingProvider):
return events
# --------------------------------------------------------------------------
# VOD
# --------------------------------------------------------------------------
def get_vod_category(self, content_id: str = "", **kwargs):
"""Delegate VOD browsing to RTLPlusVodManager."""
return self._vod_manager.get_vod_category(content_id=content_id, **kwargs)
def _parse_station_to_channel(self, station: Dict) -> Optional[StreamingChannel]:
"""
Parse a station object from RTL+ API to StreamingChannel
"""
try:
# Extract basic info
name = station.get("name", "")
channel_id = station.get("id", "")
if not name or not channel_id:
return None
# Extract logo URL
logo_url = None
if "images" in station and "alternativeLandscapeUri" in station["images"]:
logo_url = station["images"]["alternativeLandscapeUri"]
# Determine if premium channel
is_premium = station.get("isPremium", False)
# Extract watch path for potential manifest fetching
watch_path = None
if "urlData" in station and "watchPath" in station["urlData"]:
watch_path = station["urlData"]["watchPath"]
# Create channel object
channel = StreamingChannel(
name=name,
content_id=channel_id,
provider=self.provider_name,
logo_url=logo_url,
mode="live",
session_manifest=True, # RTL+ uses dynamic manifests
manifest=None, # Will be set dynamically
manifest_script=watch_path, # Store watch path for manifest fetching
content_type="LIVE",
country=self.country,
language="de",
)
# Set CDM settings for premium channels
if is_premium:
channel.use_cdm = True
channel.cdm_type = "widevine"
return channel
except Exception as e:
logger.warning(f"Error parsing station {station}: {e}")
return None
# --------------------------------------------------------------------------
# Manifest & DRM (Dispatcher)
# --------------------------------------------------------------------------
@staticmethod
def _is_vod_rrn(content_id: str) -> bool:
"""True for episode / movie / clip RRNs from the videohub namespace."""
def _is_linear_tv_channel(content_id: str) -> bool:
"""Determine if content_id refers to a linear TV channel."""
return (
content_id.startswith("rrn:watch:videohub:episode:")
or content_id.startswith("rrn:watch:videohub:movie:")
or content_id.startswith("rrn:watch:videohub:clip:")
":" not in content_id
and not content_id.startswith("rrn:")
and not content_id.startswith("/")
and not content_id.startswith("http")
)
def _fetch_manifest_data(self, content_id: str) -> Optional[list]:
"""
Fetch raw manifest data for a given content_id, with a 1-day in-memory cache.
Both get_manifest() and get_drm() call this method so the endpoint is
only hit once per content_id per day, regardless of call order.
"""
now = time.monotonic()
cached = self._manifest_cache.get(content_id)
if cached is not None:
data, ts = cached
if (now - ts) < _MANIFEST_CACHE_TTL:
logger.debug(f"RTL+ Manifest cache hit for {content_id}")
return data
manifest_url = self.rtl_config.get_manifest_url(content_id)
logger.debug(f"RTL+ Manifest Request: GET {manifest_url}")
headers = self._get_event_manifest_headers()
response = self.http_manager.get(manifest_url, operation="manifest", headers=headers)
logger.debug(f"RTL+ Manifest Response: Status={response.status_code}")
logger.debug(f"RTL+ Response Headers: {dict(response.headers)}")
response.raise_for_status()
data = response.json()
logger.debug(f"RTL+ Manifest Data: {self._sanitize_manifest_log(data)}")
self._manifest_cache[content_id] = (data, now)
return data
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
"""
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)
def _get_manifest_vod_or_event(self, content_id: str, **kwargs) -> Optional[str]:
"""Original manifest logic for VOD/events."""
try:
manifest_data = self._fetch_manifest_data(content_id)
if manifest_data is None:
@@ -379,7 +258,6 @@ class RTLPlusProvider(StreamingProvider):
logger.info(f"RTL+ Selected Manifest URL: {selected_url}")
return selected_url
# Fallback logic
for stream in manifest_data:
sources = stream.get("sources", [])
if sources:
@@ -400,45 +278,28 @@ class RTLPlusProvider(StreamingProvider):
logger.error(f"RTL+ Manifest Unexpected Error: {str(e)}")
return None
@staticmethod
def _sanitize_manifest_log(manifest_data: Dict) -> Dict:
"""
Sanitize manifest data for logging (remove sensitive information)
"""
try:
# Create a copy to avoid modifying original
sanitized = manifest_data.copy()
# Remove or truncate potentially sensitive URLs
if isinstance(sanitized, list):
for stream in sanitized:
if isinstance(stream, dict):
if "sources" in stream:
for source in stream["sources"]:
if "url" in source:
# Truncate long URLs for logging
url = source["url"]
if len(url) > 100:
source["url"] = url[:100] + "..."
return sanitized
except Exception:
return manifest_data
def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
"""
Get DRM configurations for a channel from RTL+ streaming API
Get DRM configuration for content.
For linear TV channels: uses new upfront token flow
For VOD: uses existing manifest-based DRM
"""
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)
def _get_drm_vod_or_event(self, content_id: str, **kwargs) -> List[DRMConfig]:
"""Original DRM logic for VOD/events."""
try:
manifest_data = self._fetch_manifest_data(content_id)
if manifest_data is None:
return []
drm_configs = []
# Get access token for license requests
access_token = self.authenticator.get_bearer_token()
# Look for dashhd streams (preferred quality) and extract DRM info
for stream in manifest_data:
if stream.get("name") == "dashhd" and "licenses" in stream:
licenses = stream.get("licenses", [])
@@ -448,71 +309,53 @@ class RTLPlusProvider(StreamingProvider):
if not license_url:
continue
def create_drm_config(drm_system, priority, server_url, headers):
return DRMConfig(
system=drm_system,
priority=priority,
if license_info.get("type") == "WIDEVINE":
drm_configs.append(DRMConfig(
system=DRMSystem.WIDEVINE,
priority=3,
license=LicenseConfig(
server_url=server_url,
req_headers=json.dumps(headers),
server_url=license_url,
req_headers=json.dumps(self.rtl_config.get_drm_headers(access_token)),
req_data="{CHA-RAW}",
use_http_get_request=False,
),
)
if license_info.get("type") == "WIDEVINE":
drm_configs.append(
create_drm_config(
DRMSystem.WIDEVINE,
3,
license_url,
self.rtl_config.get_drm_headers(access_token),
)
)
))
elif license_info.get("type") == "PLAYREADY":
drm_configs.append(
create_drm_config(
DRMSystem.PLAYREADY,
2,
license_url,
self.rtl_config.get_playready_drm_headers(access_token),
)
)
elif license_info.get("type") == "FAIRPLAY":
drm_configs.append(
create_drm_config(
DRMSystem.FAIRPLAY,
4,
license_url,
self.rtl_config.get_drm_headers(access_token),
)
)
drm_configs.append(DRMConfig(
system=DRMSystem.PLAYREADY,
priority=2,
license=LicenseConfig(
server_url=license_url,
req_headers=json.dumps(self.rtl_config.get_playready_drm_headers(access_token)),
req_data="{CHA-RAW}",
use_http_get_request=False,
),
))
break
return drm_configs
except requests.RequestException as e:
logger.error(f"Error fetching DRM configs for RTL+ channel {content_id}: {e}")
return []
except Exception as e:
logger.error(f"Error parsing DRM configs for RTL+ channel {content_id}: {e}")
logger.error(f"Error fetching DRM configs for RTL+ content {content_id}: {e}")
return []
@staticmethod
def get_epg_data(channel_id: str, **kwargs) -> Optional[Dict]:
"""
Get EPG data for a channel
"""
# RTL+ EPG implementation would go here
# This is a placeholder for future implementation
return None
def _fetch_manifest_data(self, content_id: str) -> Optional[list]:
"""Fetch raw manifest data for VOD/events, with cache."""
now = time.monotonic()
cached = self._manifest_cache.get(content_id)
if cached is not None:
data, ts = cached
if (now - ts) < _MANIFEST_CACHE_TTL:
logger.debug(f"RTL+ Manifest cache hit for {content_id}")
return data
def get_license_url(self, channel: StreamingChannel, **kwargs) -> Optional[str]:
"""
Get license URL for a DRM-protected channels
"""
drm_configs = self.get_drm(channel.channel_id, **kwargs)
if drm_configs:
# Return the first license URL found
return drm_configs[0].license.server_url
return None
# TODO: This legacy endpoint may eventually be migrated to Bedrock
manifest_url = f"https://stus.player.streamingtech.de/watch-playout-variants/{content_id}?platform=web"
headers = {"X-Auth-Token": self.authenticator.get_bearer_token()}
response = self.http_manager.get(manifest_url, operation="manifest", headers=headers)
response.raise_for_status()
data = response.json()
self._manifest_cache[content_id] = (data, now)
return data