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

Implement RTL Events

Browse Source
This commit is contained in:
Nirvana
2026-05-08 09:54:08 +02:00
parent 95f657393d
commit ee52b7040b
4 changed files with 574 additions and 399 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/rtlplus/event_manager.py
+269 -239
View File
@@ -3,17 +3,18 @@
RTL+ Event Manager
Handles fetching future/live events from Bedrock layout API.
Replaces legacy GraphQL-based event fetching.
Uses the homepage's "Diese Live-Events erwarten euch" block as primary source.
Follows non-sequential pagination (1 → 3 → 6 → ...) as returned by the API.
"""
import json
import re
from datetime import datetime
from typing import List, Optional, Dict
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Set
from ...base.models.event import Event, EventStatus
from ...base.utils.logger import logger
from ...base.models import DRMConfig
from .layout_helpers import unwrap_target, parse_german_datetime, extract_thumbnail
class RTLPlusEventManager:
@@ -21,23 +22,19 @@ class RTLPlusEventManager:
Manages fetching events (live streams, upcoming sports) for RTL+.
Uses the Bedrock layout API to fetch folder pages and paginated blocks.
Pagination follows the nextPage values exactly as returned by the API,
which may be non-sequential (1 → 3 → 6 → ...).
"""
# Block titles that indicate live stream content
LIVE_STREAM_BLOCK_TITLES = [
"Diese Live-Events erwarten euch", # Primary block on homepage
"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",
@@ -74,47 +71,50 @@ class RTLPlusEventManager:
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
force_refresh: bool = False,
max_pages: int = 20,
) -> List[Event]:
"""
Fetch all future/live events from the specified folder.
Fetch all future/live events.
Primary source is the homepage's "Diese Live-Events erwarten euch" block.
Falls back to specific folder if homepage fails.
Args:
folder_id: The folder ID (default "6" for Sport im Überblick)
folder_id: Fallback folder ID (default "6" for Sport)
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
max_pages: Maximum number of pagination pages to fetch
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 []
# PRIMARY: Fetch from homepage (has all events in one block)
home_events = self._fetch_events_from_homepage(force_refresh, max_pages)
if home_events:
all_events.extend(home_events)
logger.info(f"Fetched {len(home_events)} events from homepage")
else:
# FALLBACK: Fetch from specific folder
logger.debug(f"Falling back to folder {folder_id} for events")
layout = self._provider.fetch_layout(
layout_type="folder",
content_id=folder_id,
force_refresh=force_refresh,
)
if layout:
folder_events = self._extract_events_from_layout(layout)
all_events.extend(folder_events)
# 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
# Filter by time range
filtered_events = self._filter_by_time(all_events, start_time, end_time)
# Step 5: Sort chronologically
# 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")
logger.info(f"Returning {len(filtered_events)} events after filtering")
return filtered_events
def get_manifest_for_event(self, event_id: str) -> Optional[str]:
@@ -122,158 +122,192 @@ class RTLPlusEventManager:
Get manifest URL for an event.
Args:
event_id: The event folder ID (e.g., "76")
event_id: The event folder ID (e.g., "95" for Freiburg vs Braga)
Returns:
Manifest URL if available from the folder layout, None otherwise
Manifest URL if available, None otherwise
"""
layout = self._fetch_folder_layout(event_id)
layout = self._provider.fetch_layout(layout_type="folder", content_id=event_id)
if not layout:
logger.debug(f"No folder layout found for event {event_id}")
return None
# Extract video assets from the layout
# First try: Extract manifest from the folder's solo block
manifest = self._extract_manifest_from_folder_layout(layout)
if manifest:
logger.debug(f"Found manifest in folder layout for event {event_id}")
return manifest
# Second try: Extract video assets directly from layout
assets = self._provider.extract_video_assets(layout)
if assets:
manifest_url = self._provider.extract_best_manifest_url(assets)
if manifest_url:
logger.debug(f"Found manifest from assets for event {event_id}")
return manifest_url
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 manifest_url
logger.debug(f"No manifest found for event {event_id}")
return None
def get_drm_for_event(self, event_id: str) -> List[DRMConfig]:
"""
Get DRM configuration for an event.
Args:
event_id: The event folder ID
Returns:
List of DRMConfig objects
"""
layout = self._fetch_folder_layout(event_id)
layout = self._provider.fetch_layout(layout_type="folder", content_id=event_id)
if not layout:
return []
# Delegate to provider's common DRM method
return self._provider.get_drm_for_content(layout)
# --------------------------------------------------------------------------
# Internal Methods
# Homepage Event Fetching (Primary Source)
# --------------------------------------------------------------------------
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,
def _fetch_events_from_homepage(self, force_refresh: bool = False, max_pages: int = 20) -> List[Event]:
"""
Fetch events from the homepage's "Diese Live-Events erwarten euch" block.
This is the primary source because it contains ALL live events in one
paginated block, unlike folder views which only show one event.
Pagination follows the nextPage values exactly as returned by the API.
The API uses non-sequential page numbers (1 → 3 → 6 → ...).
"""
home_layout = self._provider.fetch_layout(
layout_type="alias",
content_id="home",
location=f"{self._provider.rtl_config.base_website}",
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")
if not home_layout:
logger.warning("Failed to fetch home layout for events")
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")
live_block = self._find_live_events_block(home_layout)
if not live_block:
logger.debug("No live events block found on homepage")
return []
logger.debug(f"Block {block_id}: total={total_items}, per_page={items_per_page}, next_page={next_page}")
block_id = live_block.get("id")
if not block_id:
logger.warning("Live events block missing 'id' field")
return []
# 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)
all_events: List[Event] = []
# visited_pages guards against an infinite loop if the API ever returns
# a nextPage value that points back to a page we already fetched.
visited_pages: Set[int] = set()
# 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
# Process first page (already loaded in home_layout)
initial_items = live_block.get("content", {}).get("items", [])
all_events.extend(self._extract_events_from_items(initial_items))
visited_pages.add(1)
# Get pagination info from first page
pagination = live_block.get("content", {}).get("pagination", {})
next_page = pagination.get("nextPage") # Will be 3, not 2!
pages_fetched = 1
logger.debug(f"Live events block: total={pagination.get('totalItems', 0)}, next_page={next_page}")
# Follow nextPage links exactly as returned by API.
# The API skips many page numbers (e.g. 1 → 3 → 6 → 9), so we must
# never assume the next page number — always use the value from the
# pagination response.
while next_page and next_page not in visited_pages and pages_fetched < max_pages:
visited_pages.add(next_page)
logger.debug(f"Fetching page {next_page} of live events block")
page_data = self._provider.fetch_block_page(
block_id=block_id,
page=next_page,
nb_pages=3,
service_id="rtlplus_root",
)
# 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)
all_events.extend(self._extract_events_from_items(items))
pages_fetched += 1
# 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")
logger.info(f"Fetched {len(all_events)} events from {pages_fetched} pages")
return all_events
def _find_live_events_block(self, layout: Dict) -> Optional[Dict]:
"""
Find the live events block in homepage layout.
Prefers an exact match on "Diese Live-Events erwarten euch", then
falls back to any block whose title contains a known pattern.
"""
fallback: Optional[Dict] = None
for block in layout.get("blocks", []):
if block.get("type") != "bffPaginated":
continue
block_title = (
block.get("analytics", {})
.get("tealium", {})
.get("block_title", "")
)
# Exact match — return immediately
if block_title == "Diese Live-Events erwarten euch":
logger.debug(f"Found live events block (exact): '{block_title}'")
return block
# Partial match — keep as fallback, continue looking for exact
if fallback is None:
for pattern in self.LIVE_STREAM_BLOCK_TITLES:
if pattern.lower() in block_title.lower():
logger.debug(f"Found potential live block (partial): '{block_title}'")
fallback = block
break
return fallback
# --------------------------------------------------------------------------
# Layout Extraction Helpers
# --------------------------------------------------------------------------
def _extract_events_from_layout(self, layout: Dict) -> List[Event]:
"""Extract events from any layout by scanning all bffPaginated blocks."""
all_events: List[Event] = []
for block in layout.get("blocks", []):
if block.get("type") != "bffPaginated":
continue
items = block.get("content", {}).get("items", [])
all_events.extend(self._extract_events_from_items(items))
return all_events
# --------------------------------------------------------------------------
# Event Extraction from Items
# --------------------------------------------------------------------------
def _extract_events_from_items(self, items: List[Dict]) -> List[Event]:
"""
Extract Event objects from block items.
Each item represents a live event with:
- highlight: "Fußball • Do., 07.05.26, 20:30 Uhr"
- action.target.value_layout: contains folder_id for the event
"""
events: List[Event] = []
current_time = datetime.now()
@@ -288,131 +322,144 @@ class RTLPlusEventManager:
if not highlight:
continue
# Parse the date from highlight
event_date = self._parse_date_from_highlight(highlight)
# Parse the date from highlight, falling back to description
event_date = parse_german_datetime(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
event_date = parse_german_datetime(description)
if not event_date:
continue
# Only include future events (or currently live)
# Only include future events or currently live ones
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)
# Extract title
title = item_content.get("title") or 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
# Unwrap the action target (handles lock-wrapped targets transparently)
action = item_content.get("action", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
folder_id = value_layout.get("id")
# Extract sport type
# Fallback to itemContent.id if folder_id not found
if not folder_id:
folder_id = item_content.get("id")
if not folder_id:
logger.debug(f"Skipping event with no folder_id: {title}")
continue
status = self._determine_event_status(event_date, highlight)
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,
content_id=folder_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),
logo_url=extract_thumbnail(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"),
# Store additional metadata for later manifest resolution
event.manifest_script = json.dumps({
"folder_id": folder_id,
"seo": 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]:
def _extract_manifest_from_folder_layout(self, layout: Dict) -> Optional[str]:
"""
Parse German date format from highlight string.
Extract manifest URL from a folder layout (live event detail page).
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"
The folder layout has a Solo block with a "Live ansehen" button.
The target points to a live player, which we need to fetch.
"""
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
for block in layout.get("blocks", []):
if block.get("type") != "bffPaginated":
continue
# Assume 20XX for years like '26'
if year < 100:
year = 2000 + year
for item in block.get("content", {}).get("items", []):
item_content = item.get("itemContent", {})
action = item_content.get("action", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
try:
return datetime(year, month, day, hour, minute)
except ValueError:
continue
# The target type "live" points to the player
if value_layout.get("type") == "live":
live_event_id = value_layout.get("id")
if live_event_id:
return self._get_manifest_from_live_event(live_event_id)
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()
def _get_manifest_from_live_event(self, live_event_id: str) -> Optional[str]:
"""
Get manifest from live event player page.
Args:
live_event_id: The live event ID (e.g., "rtlde_event3")
"""
live_layout = self._provider.fetch_layout(
layout_type="live",
content_id=live_event_id,
location=f"{self._provider.rtl_config.base_website}{live_event_id}",
)
if live_layout:
assets = self._provider.extract_video_assets(live_layout)
if assets:
return self._provider.extract_best_manifest_url(assets)
return None
# --------------------------------------------------------------------------
# Date / Status / Sport Helpers
# --------------------------------------------------------------------------
@staticmethod
def _is_currently_live(event_date: datetime) -> bool:
"""Check if an event is currently live (within 3 hours of start)."""
def _determine_event_status(event_date: datetime, highlight: str) -> EventStatus:
"""
Determine if event is live, upcoming, or ended.
Uses both the date/time and the presence of "Live" in the highlight text.
"""
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))
# Explicit "Live" label in the metadata takes priority
if "live" in highlight.lower():
return EventStatus.LIVE
# Event started within the last 3 hours → treat as live
if event_date <= now and event_date >= now - timedelta(hours=3):
return EventStatus.LIVE
if event_date > now:
return EventStatus.SCHEDULED
return EventStatus.ENDED
@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
import re
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()
@@ -428,26 +475,9 @@ class RTLPlusEventManager:
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
# --------------------------------------------------------------------------
# Time Filtering
# --------------------------------------------------------------------------
@staticmethod
def _filter_by_time(
@@ -461,7 +491,7 @@ class RTLPlusEventManager:
filtered = []
for event in events:
if start_time and event.end_time and event.end_time < start_time:
if start_time and event.start_time and event.start_time < start_time:
continue
if end_time and event.start_time and event.start_time > end_time:
continue
@@ -476,7 +506,7 @@ class RTLPlusEventManager:
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
cache_key = f"folder:{folder_id}:1:2"
self._provider.invalidate_layout_cache(cache_key)
else:
self._provider.invalidate_layout_cache()
lib/streaming_providers/providers/rtlplus/layout_helpers.py
+199
View File
@@ -0,0 +1,199 @@
# streaming_providers/providers/rtlplus/layout_helpers.py
"""
RTL+ Layout Helpers
Pure, stateless utilities shared across event_manager, vod_manager, and
channel_manager. No I/O, no provider references — safe to import anywhere.
Centralises:
- Lock-target unwrapping (_unwrap_target)
- German date parsing (parse_german_datetime)
- Signed image URL construction (build_image_url, extract_thumbnail)
"""
import hashlib
import re
from datetime import datetime
from typing import Dict, Optional
from .constants import RTLPlusDefaults
# ---------------------------------------------------------------------------
# Lock-target unwrapping
# ---------------------------------------------------------------------------
def unwrap_target(target: Dict) -> Dict:
"""
Unwrap a lock-wrapped action target to its original target dict.
The Bedrock API wraps pay-walled or auth-gated targets like this::
{
"type": "lock",
"value_lock": {
"originalTarget": { "type": "...", "value_layout": {...} }
}
}
Calling this function on any target — locked or not — always returns the
"real" target so callers never need the ``if target.get("type") == "lock"``
boilerplate.
Args:
target: The raw ``action.target`` dict from a layout item.
Returns:
The unwrapped target dict, or the original dict if it wasn't locked.
"""
if not isinstance(target, dict):
return {}
if target.get("type") == "lock":
lock_value = target.get("value_lock", {})
if isinstance(lock_value, dict):
inner = lock_value.get("originalTarget", {})
if isinstance(inner, dict):
return inner
return target
# ---------------------------------------------------------------------------
# German date parsing
# ---------------------------------------------------------------------------
# Ordered from most-specific to least-specific so the first match wins.
_DATE_PATTERNS = [
# "Do., 07.05.26, 20:30 Uhr" (weekday, 2-digit year, with "Uhr")
r"[A-Za-z]+\.?,\s*(\d{2})\.(\d{2})\.(\d{2}),\s*(\d{2}):(\d{2})\s*Uhr",
# "Do., 07.05.26, 20:30" (weekday, 2-digit year, without "Uhr")
r"[A-Za-z]+\.?,\s*(\d{2})\.(\d{2})\.(\d{2}),\s*(\d{2}):(\d{2})",
# "07.05.26, 20:30 Uhr" (no weekday, 2-digit year, with "Uhr")
r"(\d{2})\.(\d{2})\.(\d{2}),\s*(\d{2}):(\d{2})\s*Uhr",
# "07.05.2026, 20:30" (no weekday, 4-digit year)
r"(\d{2})\.(\d{2})\.(\d{4}),\s*(\d{2}):(\d{2})",
# "07.05.26 20:30" (no comma)
r"(\d{2})\.(\d{2})\.(\d{2})\s+(\d{2}):(\d{2})",
# "07.05.26" (date only, no time)
r"(\d{2})\.(\d{2})\.(\d{2,4})",
]
def parse_german_datetime(text: str) -> Optional[datetime]:
"""
Parse a German-formatted date/time string embedded in arbitrary text.
Handles all variants seen in RTL+ Bedrock highlight strings, e.g.::
"Fußball • Do., 07.05.26, 20:30 Uhr"
"Motorsport • Fr., 15.05.26, 13:10 Uhr"
"Motorsport \u2022 Sa., 16.05.26, 14:15 Uhr"
"07.05.2026, 20:30"
Two-digit years are expanded to the 21st century (``26`` → ``2026``).
Args:
text: Any string that may contain a German date.
Returns:
A ``datetime`` object, or ``None`` if no recognisable date was found.
"""
for pattern in _DATE_PATTERNS:
match = re.search(pattern, text)
if not match:
continue
groups = match.groups()
if len(groups) < 3:
continue
try:
day = int(groups[0])
month = int(groups[1])
year = int(groups[2])
# Expand 2-digit year
if year < 100:
year = 2000 + year
hour = int(groups[3]) if len(groups) > 3 else 0
minute = int(groups[4]) if len(groups) > 4 else 0
return datetime(year, month, day, hour, minute)
except ValueError:
continue # Try the next pattern
return None
# ---------------------------------------------------------------------------
# Signed image URL construction
# ---------------------------------------------------------------------------
def build_image_url(image_id: str) -> str:
"""
Build a signed Bedrock CDN image URL for the given image ID.
Hash formula: ``SHA1("/v2/images/{id}/raw?{params}" + IMAGE_SIGNING_KEY)``
The signing key is appended as a suffix (secret-suffix construction),
verified against multiple known-good URLs captured from network traces.
Args:
image_id: The raw image ID returned by the Bedrock API.
Returns:
A fully-qualified, signed image URL.
"""
suffix = f"/{image_id}/raw?{RTLPlusDefaults.IMAGE_PARAMS}"
signed_path = f"/v2/images{suffix}"
image_hash = hashlib.sha1(
(signed_path + RTLPlusDefaults.IMAGE_SIGNING_KEY).encode()
).hexdigest()
return f"{RTLPlusDefaults.IMAGE_BASE_URL}{suffix}&hash={image_hash}"
def extract_thumbnail(item_content: Dict) -> Optional[str]:
"""
Extract the best available thumbnail URL from a Bedrock item's content dict.
Tries aspect ratios in preference order: 16:9, 3:1, 1:1, 2:3.
Falls back to the top-level ``image.id`` if no ratio-keyed ID is found.
Args:
item_content: The ``itemContent`` dict from a Bedrock block item.
Returns:
A signed CDN URL, or ``None`` if no image data is present.
"""
image = item_content.get("image", {})
if not image:
return None
for ratio in ("16:9", "3:1", "1:1", "2:3"):
image_id = image.get("idsByRatio", {}).get(ratio)
if image_id:
return build_image_url(image_id)
image_id = image.get("id")
if image_id:
return build_image_url(image_id)
return None
def extract_thumbnail_from_layout(layout: Dict) -> Optional[str]:
"""
Extract a thumbnail URL from a top-level program/video layout dict.
Reads from ``layout.seo.image.id``, which is populated on program and
video layout responses but not on individual block items.
Args:
layout: A full Bedrock layout response dict.
Returns:
A signed CDN URL, or ``None`` if no SEO image is present.
"""
image_id = layout.get("seo", {}).get("image", {}).get("id")
if image_id:
return build_image_url(image_id)
return None
lib/streaming_providers/providers/rtlplus/provider.py
+62 -59
View File
@@ -10,6 +10,7 @@ from ...base.provider import StreamingProvider
from ...base.utils import logger
from .auth import RTLPlusAuthenticator
from .constants import RTLPlusConfig, RTLPlusDefaults
from .layout_helpers import unwrap_target
from .vod_manager import RTLPlusVodManager
from .channel_manager import RTLPlusChannelManager
from .event_manager import RTLPlusEventManager
@@ -117,7 +118,7 @@ class RTLPlusProvider(StreamingProvider):
force_refresh: bool = False,
) -> Optional[Dict]:
"""
Fetch any layout (live/video/folder/program/block) with caching.
Fetch any layout (live/video/folder/program/block/alias) with caching.
"""
if block_page is None:
block_page = RTLPlusDefaults.DEFAULT_BLOCK_PAGE
@@ -191,6 +192,63 @@ class RTLPlusProvider(StreamingProvider):
logger.error(f"Failed to fetch {layout_type} layout for {clean_content_id}: {e}")
return None
def fetch_block_page(
self,
block_id: str,
page: int,
nb_pages: int = 3,
service_id: str = "rtlplus_root",
) -> Optional[Dict]:
"""
Fetch a specific page of a service block from the Bedrock API.
URL pattern: /service/{service_id}/block/{block_id}?nbPages={nb_pages}&page={page}
This is intentionally separate from ``fetch_layout`` because the service/block
endpoint uses different URL structure and query parameters (``page`` / ``nbPages``
instead of ``blockPage`` / ``nbPages``), and its responses are not cached — the
same page number may return different content as events are added or removed.
Args:
block_id: Full block ID as returned by the layout API, e.g.
``"page_69fcecf6347a02.20778370--6294a9fc-55a7-43cf-aa18-e27f1bd1a2c3"``
page: Page number to fetch. May be non-sequential (1 → 3 → 6 → …)
as dictated by the API's ``pagination.nextPage`` field.
nb_pages: Number of pages to request per call (default 3, matching API default).
service_id: Bedrock service identifier (default ``"rtlplus_root"``).
Returns:
Parsed JSON response dict, or ``None`` on failure.
"""
url = f"{self.rtl_config.bedrock_layout_base}/service/{service_id}/block/{block_id}"
params = {"nbPages": nb_pages, "page": page}
oauth_token = self.get_user_bearer_token() or self.authenticator.get_bearer_token()
if not oauth_token:
logger.error("No OAuth token available for fetch_block_page")
return None
try:
bedrock_token = self.authenticator.get_bedrock_token()
except Exception as e:
logger.error(f"Failed to get Bedrock token for fetch_block_page: {e}")
return None
if not bedrock_token:
logger.error("No Bedrock token available for fetch_block_page")
return None
location = f"{self.rtl_config.base_website}"
headers = self.rtl_config.get_layout_headers(oauth_token, bedrock_token, location)
try:
response = self.http_manager.get(url, headers=headers, params=params, operation="api")
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Failed to fetch block page {page} for {block_id}: {e}")
return None
@staticmethod
def extract_video_assets(layout_data: Dict) -> List[Dict]:
"""
@@ -401,7 +459,7 @@ class RTLPlusProvider(StreamingProvider):
# If content_id starts with program_, we need to find the clip_id from the program
if content_id.startswith("program_"):
program_id = content_id[8:] # Remove "program_" prefix
layout = self._fetch_program_layout(program_id)
layout = self.fetch_layout(layout_type="program", content_id=program_id)
if layout:
clip_id = self._find_clip_id_in_program_layout(layout)
if clip_id:
@@ -426,37 +484,6 @@ class RTLPlusProvider(StreamingProvider):
# Fall back to VOD/event manifest extraction
return self._get_manifest_vod_or_event(content_id, **kwargs)
def _fetch_program_layout(self, program_id: str) -> Optional[Dict]:
"""Fetch a program layout by program ID."""
oauth_token = self.get_user_bearer_token()
if not oauth_token:
logger.error("No user authentication for program layout")
return None
try:
bedrock_token = self.authenticator.get_bedrock_token()
if not bedrock_token:
logger.error("Failed to get Bedrock token")
return None
except Exception as e:
logger.error(f"Failed to get Bedrock token: {e}")
return None
clean_id = program_id.replace("program_", "")
url = f"{self.rtl_config.bedrock_layout_base}/program/{clean_id}/layout"
location = f"{self.rtl_config.base_website}p_{clean_id}-p_{clean_id}"
headers = self.rtl_config.get_layout_headers(oauth_token, bedrock_token, location)
params = {"blockPage": 1, "nbPages": 2}
try:
response = self.http_manager.get(url, headers=headers, params=params, operation="api")
response.raise_for_status()
return response.json()
except Exception as e:
logger.error(f"Failed to fetch program layout for {program_id}: {e}")
return None
def _find_clip_id_in_program_layout(self, layout: Dict) -> Optional[str]:
"""Extract the clip_id from a program layout."""
if not layout:
@@ -485,11 +512,7 @@ class RTLPlusProvider(StreamingProvider):
item_content = item.get("itemContent", {})
action = item_content.get("action", {})
target = action.get("target", {})
if target.get("type") == "lock":
target = target.get("value_lock", {}).get("originalTarget", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
if value_layout.get("type") == "video":
return value_layout.get("id")
@@ -509,7 +532,7 @@ class RTLPlusProvider(StreamingProvider):
# If content_id starts with program_, find the clip_id
if content_id.startswith("program_"):
program_id = content_id[8:]
layout = self._fetch_program_layout(program_id)
layout = self.fetch_layout(layout_type="program", content_id=program_id)
if layout:
clip_id = self._find_clip_id_in_program_layout(layout)
if clip_id:
@@ -692,26 +715,6 @@ class RTLPlusProvider(StreamingProvider):
logger.error(f"Failed to get DRM: {e}")
return []
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:
return data
# 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
def get_user_bearer_token(self) -> Optional[str]:
"""Get a user-authenticated bearer token, upgrading if necessary. Returns None if impossible."""
from ...base.auth.base_auth import TokenAuthLevel
lib/streaming_providers/providers/rtlplus/vod_manager.py
+44 -101
View File
@@ -20,6 +20,13 @@ from ...base.models.vod import VodCategory, VodItem
from ...base.models import DRMConfig
from ...base.utils.logger import logger
from .constants import RTLPlusDefaults
from .layout_helpers import (
unwrap_target,
parse_german_datetime,
build_image_url,
extract_thumbnail,
extract_thumbnail_from_layout,
)
FOLDER_NAMES = {
"4": "Filme",
@@ -323,12 +330,7 @@ class RTLPlusVodManager:
item_content = item.get("itemContent", {})
action = item_content.get("action", {})
target = action.get("target", {})
# Handle lock-wrapped targets
if target.get("type") == "lock":
target = target.get("value_lock", {}).get("originalTarget", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
if value_layout.get("type") == "video" and value_layout.get("id") == clip_id:
@@ -534,7 +536,7 @@ class RTLPlusVodManager:
if layout_title and layout_title != movie_item.name:
movie_item.name = layout_title
if not movie_item.logo_url:
movie_item.logo_url = self._extract_thumbnail_from_layout(layout)
movie_item.logo_url = extract_thumbnail_from_layout(layout)
if not movie_item.description:
movie_item.description = (
layout.get("entity", {}).get("metadata", {}).get("description")
@@ -791,22 +793,15 @@ class RTLPlusVodManager:
# Check for video in action target
action = item_content.get("action", {})
if isinstance(action, dict):
target = action.get("target", {})
if isinstance(target, dict):
# Handle lock-wrapped targets
if target.get("type") == "lock":
lock_value = target.get("value_lock", {})
if isinstance(lock_value, dict):
target = lock_value.get("originalTarget", {})
value_layout = target.get("value_layout", {})
if isinstance(value_layout, dict) and value_layout.get("type") == "video":
vod_item = self._extract_vod_item_from_block_item(item)
if vod_item:
clip_id = value_layout.get("id")
if clip_id:
vod_item.content_id = clip_id
return vod_item
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
if isinstance(value_layout, dict) and value_layout.get("type") == "video":
vod_item = self._extract_vod_item_from_block_item(item)
if vod_item:
clip_id = value_layout.get("id")
if clip_id:
vod_item.content_id = clip_id
return vod_item
# Check direct itemContent type
if item_content.get("type") == "video":
@@ -820,7 +815,7 @@ class RTLPlusVodManager:
episode_number=-1,
)
vod_item.description = item_content.get("description")
vod_item.logo_url = self._extract_thumbnail(item_content)
vod_item.logo_url = extract_thumbnail(item_content)
return vod_item
return None
@@ -944,7 +939,7 @@ class RTLPlusVodManager:
episode_number=video_meta.get("episode", -1),
)
vod_item.description = metadata.get("description")
vod_item.logo_url = self._extract_thumbnail_from_layout(layout)
vod_item.logo_url = extract_thumbnail_from_layout(layout)
vod_item.duration_seconds = video_meta.get("duration")
vod_item.genre = parent.get("seo", "")
vod_item.series_title = parent.get("name")
@@ -977,16 +972,11 @@ class RTLPlusVodManager:
logger.warning(f"_extract_vod_item_from_block_item: item_content is {type(item_content)}")
return None
# Try to get video reference from action target first
# Unwrap lock-wrapped targets before reading the action target
action = item_content.get("action", {})
target = action.get("target", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
# Also check for lock-wrapped targets
if target.get("type") == "lock":
target = target.get("value_lock", {}).get("originalTarget", {})
value_layout = target.get("value_layout", {})
clip_id = None
program_id = None
program_slug = None
@@ -1004,16 +994,15 @@ class RTLPlusVodManager:
for action_key in ("onClickAction", "primaryAction", "secondaryAction"):
alt_action = item_content.get(action_key, {})
if isinstance(alt_action, dict):
alt_target = alt_action.get("target", {})
if isinstance(alt_target, dict):
alt_value = alt_target.get("value_layout", {})
if isinstance(alt_value, dict) and alt_value.get("type") == "video":
clip_id = alt_value.get("id")
parent = alt_value.get("parent", {})
program_id = parent.get("id")
program_slug = parent.get("seo")
if clip_id:
break
alt_target = unwrap_target(alt_action.get("target", {}))
alt_value = alt_target.get("value_layout", {})
if isinstance(alt_value, dict) and alt_value.get("type") == "video":
clip_id = alt_value.get("id")
parent = alt_value.get("parent", {})
program_id = parent.get("id")
program_slug = parent.get("seo")
if clip_id:
break
if not clip_id:
return None
@@ -1064,7 +1053,7 @@ class RTLPlusVodManager:
episode_number=episode_number if episode_number is not None else -1,
)
vod_item.description = item_content.get("description")
vod_item.logo_url = self._extract_thumbnail(item_content)
vod_item.logo_url = extract_thumbnail(item_content)
vod_item.duration_seconds = self._extract_duration(item_content)
vod_item.progress = item_content.get("progress", 0)
@@ -1088,12 +1077,7 @@ class RTLPlusVodManager:
return None
action = item_content.get("action", {})
target = action.get("target", {})
# Handle lock-wrapped targets
if target.get("type") == "lock":
target = target.get("value_lock", {}).get("originalTarget", {})
target = unwrap_target(action.get("target", {}))
value_layout = target.get("value_layout", {})
layout_type = value_layout.get("type")
@@ -1132,7 +1116,7 @@ class RTLPlusVodManager:
name=name,
content_id=content_id,
provider=self._provider.provider_name,
logo_url=self._extract_thumbnail(item_content),
logo_url=extract_thumbnail(item_content),
description=item_content.get("description") or item_content.get("highlight"),
)
@@ -1149,45 +1133,21 @@ class RTLPlusVodManager:
# Thumbnail / duration helpers (pure, no I/O)
# ------------------------------------------------------------------
@staticmethod
def _build_image_url(image_id: str) -> str:
"""Build a signed Bedrock CDN image URL.
# build_image_url and extract_thumbnail are re-exported here as classmethods
# for any callers that still reference them via the class (e.g. tests).
# New code should import directly from layout_helpers.
Hash formula: SHA1("/v2/images/{id}/raw?{params}" + IMAGE_SIGNING_KEY)
Note: path_and_query starts with /{id}/raw since IMAGE_BASE_URL already
contains /v2/images — the full signed path is /v2/images/{id}/raw?{params}.
This is a keyed hash (secret-suffix construction), verified against
three known-good URLs from network traces.
"""
import hashlib
from .constants import RTLPlusDefaults
suffix = f"/{image_id}/raw?{RTLPlusDefaults.IMAGE_PARAMS}"
signed_path = f"/v2/images{suffix}"
image_hash = hashlib.sha1(
(signed_path + RTLPlusDefaults.IMAGE_SIGNING_KEY).encode()
).hexdigest()
return f"{RTLPlusDefaults.IMAGE_BASE_URL}{suffix}&hash={image_hash}"
@classmethod
def _build_image_url(cls, image_id: str) -> str:
return build_image_url(image_id)
@classmethod
def _extract_thumbnail(cls, item_content: Dict) -> Optional[str]:
image = item_content.get("image", {})
if not image:
return None
for ratio in ("16:9", "3:1", "1:1", "2:3"):
image_id = image.get("idsByRatio", {}).get(ratio)
if image_id:
return cls._build_image_url(image_id)
image_id = image.get("id")
if image_id:
return cls._build_image_url(image_id)
return None
return extract_thumbnail(item_content)
@classmethod
def _extract_thumbnail_from_layout(cls, layout: Dict) -> Optional[str]:
image_id = layout.get("seo", {}).get("image", {}).get("id")
if image_id:
return cls._build_image_url(image_id)
return None
return extract_thumbnail_from_layout(layout)
@staticmethod
def _extract_duration(item_content: Dict) -> Optional[int]:
@@ -1208,25 +1168,8 @@ class RTLPlusVodManager:
highlight = item_content.get("highlight", "")
if not highlight:
return False
date_patterns = [
r"(\d{2})\.(\d{2})\.(\d{2}),?\s*(\d{2}):(\d{2})", # DD.MM.YY, HH:MM
r"(\d{2})\.(\d{2})\.(\d{2})\s+(\d{2}):(\d{2})", # DD.MM.YY HH:MM
]
for pattern in date_patterns:
match = re.search(pattern, highlight)
if match:
day, month, year, hour, minute = map(int, match.groups())
if year < 100:
year = 2000 + year
try:
event_date = datetime(year, month, day, hour, minute)
if event_date > datetime.now():
return True
except ValueError:
continue
return False
event_date = parse_german_datetime(highlight)
return event_date is not None and event_date > datetime.now()
@staticmethod
def _is_event_item(item_content: Dict) -> bool: