From 0171c5366b6c12b6a6247efd475195319560e4d2 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Sat, 6 Dec 2025 16:13:27 +0100 Subject: [PATCH] Improve EPG --- .../base/epg/epg_manager.py | 64 ++++++++- .../base/epg/epg_parser.py | 109 +++++++++++++-- .../base/models/epg_models.py | 130 +++++++++++++++++- 3 files changed, 285 insertions(+), 18 deletions(-) diff --git a/lib/streaming_providers/base/epg/epg_manager.py b/lib/streaming_providers/base/epg/epg_manager.py index 679756a..db0f1ab 100644 --- a/lib/streaming_providers/base/epg/epg_manager.py +++ b/lib/streaming_providers/base/epg/epg_manager.py @@ -10,6 +10,29 @@ Orchestrates cache, mapping, and parsing components Refactored to use EPGEntry objects internally while maintaining backward-compatible dictionary output for external consumers. + +PROVIDER ENCODING IN BROADCAST IDs: +------------------------------------ +The EPGManager now encodes provider information into broadcast IDs, +enabling provider identification during catchup operations: + +Example workflow: + # 1. EPG is fetched and parsed (provider is known) + manager = EPGManager() + epg_data = manager.get_epg("rtlplus", "rtl") + # broadcast_id now contains encoded "rtlplus" provider hash + + # 2. Later, in catchup handler (only broadcast_id available) + def get_catchup_stream(broadcast_id): + # Identify provider from broadcast_id + provider = manager.get_provider_from_broadcast_id(broadcast_id) + # -> "rtlplus" + + # Now we can call the right provider's catchup API + return provider_registry[provider].get_catchup(broadcast_id) + +This solves the problem of needing provider information for catchup +when only the broadcast_id is passed from Kodi. """ from datetime import datetime, timedelta @@ -146,11 +169,13 @@ class EPGManager: logger.debug(f"EPGManager: Time range: {start_ts} to {end_ts}") # Step 4: Parse EPG for channel and time range (returns EPGEntry objects) + # Pass provider_name to enable provider encoding in broadcast_id epg_entries: List[EPGEntry] = self.parser.parse_epg_for_channel( xml_path, epg_channel_id, start_ts, - end_ts + end_ts, + provider_name # Enable provider encoding ) logger.info( @@ -213,7 +238,8 @@ class EPGManager: xml_path, epg_channel_id, start_ts, - end_ts + end_ts, + provider_name # Enable provider encoding ) except Exception as e: @@ -312,3 +338,37 @@ class EPGManager: True if mapping exists """ return self.mapping.has_mapping(provider_name, channel_id) + + def get_provider_from_broadcast_id(self, broadcast_id: int) -> Optional[str]: + """ + Get provider name from an encoded broadcast ID. + Useful for catchup operations where only broadcast_id is available. + + Args: + broadcast_id: Encoded broadcast ID from EPG entry + + Returns: + Provider name, or None if not found + + Example: + # In catchup handler + provider = manager.get_provider_from_broadcast_id(broadcast_id) + if provider: + # Get catchup stream from provider + stream_url = get_catchup_stream(provider, broadcast_id) + """ + return self.parser.get_provider_from_broadcast_id(broadcast_id) + + @staticmethod + def verify_broadcast_id_provider(broadcast_id: int, provider_name: str) -> bool: + """ + Verify if a broadcast ID belongs to a specific provider. + + Args: + broadcast_id: Encoded broadcast ID + provider_name: Provider name to verify + + Returns: + True if broadcast_id was generated for this provider + """ + return EPGEntry.verify_provider(broadcast_id, provider_name) diff --git a/lib/streaming_providers/base/epg/epg_parser.py b/lib/streaming_providers/base/epg/epg_parser.py index 7303ca1..f282bf0 100644 --- a/lib/streaming_providers/base/epg/epg_parser.py +++ b/lib/streaming_providers/base/epg/epg_parser.py @@ -7,7 +7,6 @@ Parses XMLTV format and converts to Kodi PVR EPG format import xml.etree.ElementTree as ET import gzip -import hashlib from datetime import datetime from typing import List, Optional, Tuple, Any, Dict from ..models.epg_models import EPGEntry @@ -27,6 +26,54 @@ class EPGParser: Now returns EPGEntry objects internally for validation and type safety. """ + def __init__(self): + """Initialize parser with provider registry for broadcast ID encoding.""" + # Provider registry maps provider hash -> provider name + # Used to look up provider from broadcast_id during catchup + self._provider_registry: Dict[int, str] = {} + + def register_provider(self, provider_name: str) -> int: + """ + Register a provider and return its hash. + This enables provider lookup from broadcast IDs. + + Args: + provider_name: Provider name to register + + Returns: + Provider hash (16-bit) + """ + import hashlib + provider_hash_obj = hashlib.sha256(provider_name.encode('utf-8')) + provider_hash = int(provider_hash_obj.hexdigest()[:4], 16) + + self._provider_registry[provider_hash] = provider_name + logger.debug(f"Registered provider '{provider_name}' with hash {provider_hash:04x}") + + return provider_hash + + def get_provider_from_broadcast_id(self, broadcast_id: int) -> Optional[str]: + """ + Get provider name from broadcast ID. + Useful for catchup operations where only broadcast_id is available. + + Args: + broadcast_id: Encoded broadcast ID + + Returns: + Provider name, or None if not found in registry + """ + provider_hash = EPGEntry.get_provider_hash(broadcast_id) + provider_name = self._provider_registry.get(provider_hash) + + if not provider_name: + logger.warning( + f"Provider not found for broadcast_id {broadcast_id} " + f"(hash: {provider_hash:04x})" + ) + + return provider_name + @staticmethod def parse_xmltv_time(time_str: str) -> Optional[int]: """ @@ -72,22 +119,35 @@ class EPGParser: return None @staticmethod - def generate_broadcast_id(channel_id: str, start_time: int) -> int: + def generate_broadcast_id( + channel_id: str, + start_time: int, + provider_name: Optional[str] = None + ) -> int: """ Generate deterministic broadcast ID from channel and start time. + If provider_name is given, uses encode_broadcast_id() for provider-aware IDs. + Otherwise, falls back to legacy hash-only method. + Args: channel_id: EPG channel ID start_time: Unix timestamp of programme start + provider_name: Optional provider name for encoding Returns: Unique integer broadcast ID """ - # Create hash from channel_id and start_time - hash_input = f"{channel_id}_{start_time}".encode('utf-8') - hash_digest = hashlib.sha256(hash_input).hexdigest() - # Convert first 8 hex chars to int (32-bit) - return int(hash_digest[:8], 16) + if provider_name: + # New method: encode provider information + return EPGEntry.encode_broadcast_id(provider_name, channel_id, start_time) + else: + # Legacy method: simple hash (for backward compatibility) + import hashlib + hash_input = f"{channel_id}_{start_time}".encode('utf-8') + hash_digest = hashlib.sha256(hash_input).hexdigest() + # Convert first 8 hex chars to int (32-bit) + return int(hash_digest[:8], 16) @staticmethod def parse_episode_num(episode_elem: ET.Element) -> Tuple[Optional[int], Optional[int], Optional[int]]: @@ -229,14 +289,19 @@ class EPGParser: else: return EPGGenre.UNDEFINED - @staticmethod - def parse_programme(programme_elem: ET.Element, epg_channel_id: str) -> Optional[EPGEntry]: + def parse_programme( + self, + programme_elem: ET.Element, + epg_channel_id: str, + provider_name: Optional[str] = None + ) -> Optional[EPGEntry]: """ Parse a single XMLTV programme element to EPGEntry object. Args: programme_elem: XML programme element epg_channel_id: EPG channel ID for validation + provider_name: Optional provider name for encoding in broadcast_id Returns: EPGEntry object, or None if required fields missing or validation fails @@ -261,8 +326,16 @@ class EPGParser: title_elem = programme_elem.find('title') title = title_elem.text if title_elem is not None and title_elem.text else "Unknown" - # Generate broadcast ID - broadcast_id = EPGParser.generate_broadcast_id(epg_channel_id, start_time) + # Generate broadcast ID (with provider encoding if available) + broadcast_id = EPGParser.generate_broadcast_id( + epg_channel_id, + start_time, + provider_name + ) + + # Register provider if given + if provider_name and hasattr(self, '_provider_registry'): + self.register_provider(provider_name) # Build EPGEntry with required fields entry_kwargs: Dict[str, Any] = { @@ -411,12 +484,13 @@ class EPGParser: logger.debug(f"Opening plain EPG file: {file_path}") return open(file_path, 'r', encoding='utf-8') - @staticmethod def parse_epg_for_channel( + self, xml_path: str, epg_channel_id: str, start_time: Optional[int] = None, - end_time: Optional[int] = None + end_time: Optional[int] = None, + provider_name: Optional[str] = None ) -> List[EPGEntry]: """ Parse EPG data for a specific channel within a time range. @@ -427,6 +501,7 @@ class EPGParser: epg_channel_id: EPG channel ID to filter for start_time: Start of time range (Unix timestamp), None for no lower bound end_time: End of time range (Unix timestamp), None for no upper bound + provider_name: Optional provider name for encoding in broadcast_id Returns: List of EPGEntry objects @@ -444,8 +519,12 @@ class EPGParser: if elem.tag == 'programme': # Check if this programme matches our channel if elem.get('channel') == epg_channel_id: - # Parse the programme - epg_entry = EPGParser.parse_programme(elem, epg_channel_id) + # Parse the programme with provider info + epg_entry = self.parse_programme( + elem, + epg_channel_id, + provider_name + ) if epg_entry: # Filter by time range using EPGEntry methods diff --git a/lib/streaming_providers/base/models/epg_models.py b/lib/streaming_providers/base/models/epg_models.py index b1d079c..bf4cfc8 100644 --- a/lib/streaming_providers/base/models/epg_models.py +++ b/lib/streaming_providers/base/models/epg_models.py @@ -5,7 +5,7 @@ EPG Models - Data classes for Electronic Program Guide entries Based on Kodi PVR EPG Tag specification (ETSI EN 300 468 DVB-SI standard) """ -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Optional, List from datetime import datetime from enum import IntEnum @@ -44,6 +44,29 @@ class EPGEntry: This class matches the dictionary format used by epg_parser.py and expected by the C++ PVR frontend. + + BROADCAST ID ENCODING: + ---------------------- + The broadcast_id field uses a special encoding scheme that embeds provider + information for catchup functionality: + + - Bits 0-15 (lower 16 bits): Provider hash + - Bits 16-31 (upper 16 bits): Event hash (channel + start time) + + This allows identifying the provider from just the broadcast_id, which is + critical for catchup operations where only the broadcast_id is available. + + Usage: + # Creating entries (done by parser) + broadcast_id = EPGEntry.encode_broadcast_id("rtlplus", "rtl", start_time) + + # In catchup handler (only broadcast_id available) + provider_hash = EPGEntry.get_provider_hash(broadcast_id) + # Look up provider from hash in registry + + # Or verify provider + if EPGEntry.verify_provider(broadcast_id, "rtlplus"): + # Get catchup stream from rtlplus """ # Required fields @@ -366,6 +389,111 @@ class EPGEntry: return [] return [item.strip() for item in text.split(EPG_STRING_TOKEN_SEPARATOR) if item.strip()] + @staticmethod + def encode_broadcast_id( + provider_name: str, + channel_id: str, + start_time: int + ) -> int: + """ + Generate deterministic broadcast ID with encoded provider information. + + The ID structure allows extraction of provider hash for catchup operations: + - Bits 0-15 (lower 16 bits): Provider hash (65536 possible values) + - Bits 16-31 (upper 16 bits): Event hash (channel + start time) + + This ensures: + 1. Unique IDs across different providers + 2. Provider can be identified from broadcast_id alone + 3. Same event on same provider always gets same ID + + Args: + provider_name: Provider name (e.g., "rtlplus", "joyn_de") + channel_id: Channel ID (e.g., "rtl", "prosieben") + start_time: Unix timestamp of programme start + + Returns: + Unique 32-bit broadcast ID with encoded provider info + + Example: + broadcast_id = EPGEntry.encode_broadcast_id("rtlplus", "rtl", 1234567890) + provider_hash = EPGEntry.get_provider_hash(broadcast_id) + """ + import hashlib + + # Generate provider hash (16 bits) + provider_hash_obj = hashlib.sha256(provider_name.encode('utf-8')) + provider_hash = int(provider_hash_obj.hexdigest()[:4], 16) # 16 bits + + # Generate event hash from channel + start time (16 bits) + event_input = f"{channel_id}_{start_time}".encode('utf-8') + event_hash_obj = hashlib.sha256(event_input) + event_hash = int(event_hash_obj.hexdigest()[:4], 16) # 16 bits + + # Combine: upper 16 bits = event hash, lower 16 bits = provider hash + broadcast_id = (event_hash << 16) | provider_hash + + # Ensure positive and non-zero + return broadcast_id if broadcast_id > 0 else 1 + + @staticmethod + def get_provider_hash(broadcast_id: int) -> int: + """ + Extract provider hash from encoded broadcast ID. + + Args: + broadcast_id: Encoded broadcast ID from encode_broadcast_id() + + Returns: + 16-bit provider hash (lower 16 bits of broadcast_id) + + Example: + broadcast_id = EPGEntry.encode_broadcast_id("rtlplus", "rtl", 1234567890) + provider_hash = EPGEntry.get_provider_hash(broadcast_id) + # Use provider_hash to look up provider from registry + """ + return broadcast_id & 0xFFFF + + @staticmethod + def get_event_hash(broadcast_id: int) -> int: + """ + Extract event hash from encoded broadcast ID. + + Args: + broadcast_id: Encoded broadcast ID from encode_broadcast_id() + + Returns: + 16-bit event hash (upper 16 bits of broadcast_id) + """ + return (broadcast_id >> 16) & 0xFFFF + + @staticmethod + def verify_provider(broadcast_id: int, provider_name: str) -> bool: + """ + Verify if a broadcast ID matches a given provider. + + Args: + broadcast_id: Encoded broadcast ID + provider_name: Provider name to verify against + + Returns: + True if broadcast_id was generated for this provider + + Example: + if EPGEntry.verify_provider(broadcast_id, "rtlplus"): + print("This EPG entry is from rtlplus") + """ + import hashlib + + # Get provider hash from broadcast_id + stored_hash = EPGEntry.get_provider_hash(broadcast_id) + + # Calculate hash for given provider name + provider_hash_obj = hashlib.sha256(provider_name.encode('utf-8')) + calculated_hash = int(provider_hash_obj.hexdigest()[:4], 16) + + return stored_hash == calculated_hash + def __post_init__(self): """Validate required fields after initialization.""" if self.broadcast_id <= EPG_TAG_INVALID_UID: