diff --git a/lib/streaming_providers/base/epg/__init__.py b/lib/streaming_providers/base/epg/__init__.py new file mode 100644 index 0000000..b2ff0a1 --- /dev/null +++ b/lib/streaming_providers/base/epg/__init__.py @@ -0,0 +1,17 @@ +#!/usr/bin/env python3 +# streaming_providers/base/epg/__init__.py +""" +EPG Module for XMLTV EPG handling +""" + +from .epg_manager import EPGManager +from .epg_cache import EPGCache +from .epg_mapping import EPGMapping +from .epg_parser import EPGParser + +__all__ = [ + 'EPGManager', + 'EPGCache', + 'EPGMapping', + 'EPGParser' +] \ No newline at end of file diff --git a/lib/streaming_providers/base/epg/epg_cache.py b/lib/streaming_providers/base/epg/epg_cache.py new file mode 100644 index 0000000..e69de29 diff --git a/lib/streaming_providers/base/epg/epg_manager.py b/lib/streaming_providers/base/epg/epg_manager.py new file mode 100644 index 0000000..72300c5 --- /dev/null +++ b/lib/streaming_providers/base/epg/epg_manager.py @@ -0,0 +1,274 @@ +#!/usr/bin/env python3 +# streaming_providers/base/epg/epg_manager.py +""" +EPG Manager - Central coordinator for EPG operations +Orchestrates cache, mapping, and parsing components +""" + +from datetime import datetime, timedelta +from typing import List, Dict, Optional +from .epg_cache import EPGCache +from .epg_mapping import EPGMapping +from .epg_parser import EPGParser +from ..utils.logger import logger + + +class EPGManager: + """ + Central manager for EPG operations. + Coordinates cache management, channel mapping, and EPG parsing. + """ + + # Default EPG source URL (can be overridden by addon setting) + DEFAULT_EPG_URL = "https://example.com/epg.xml.gz" + + def __init__( + self, + addon_path: Optional[str] = None, + user_data_path: Optional[str] = None, + epg_url: Optional[str] = None + ): + """ + Initialize EPG manager with all components. + + Args: + addon_path: Path to addon root (for default mapping) + user_data_path: Path to addon data directory (for user mapping and cache) + epg_url: EPG source URL (defaults to setting or DEFAULT_EPG_URL) + """ + # Get paths if not provided + if addon_path is None or user_data_path is None: + addon_path, user_data_path = self._get_default_paths() + + # Initialize components + self.cache = EPGCache(vfs_subdir="epg_cache") + self.mapping = EPGMapping(addon_path, user_data_path) + self.parser = EPGParser() + + # Get EPG URL from settings or use default + self.epg_url = epg_url or self._get_epg_url_from_settings() or self.DEFAULT_EPG_URL + + logger.info(f"EPGManager: Initialized with EPG URL: {self.epg_url}") + + @staticmethod + def _get_default_paths() -> tuple: + """ + Get default paths for addon and user data. + + Returns: + Tuple of (addon_path, user_data_path) + """ + try: + import xbmcaddon + import xbmcvfs + + addon = xbmcaddon.Addon() + addon_path = addon.getAddonInfo('path') + user_data_path = xbmcvfs.translatePath( + f"special://userdata/addon_data/{addon.getAddonInfo('id')}" + ) + + return addon_path, user_data_path + + except Exception as e: + logger.warning(f"EPGManager: Could not get Kodi paths: {e}, using fallback") + # Fallback for testing outside Kodi + import os + return os.getcwd(), os.path.join(os.getcwd(), 'userdata') + + @staticmethod + def _get_epg_url_from_settings() -> Optional[str]: + """ + Get EPG URL from addon settings. + + Returns: + URL from settings, or None if not set + """ + try: + import xbmcaddon + addon = xbmcaddon.Addon() + url = addon.getSetting('epg_xml_url') + + if url and url.strip(): + logger.info(f"EPGManager: Using EPG URL from settings: {url}") + return url.strip() + + except Exception as e: + logger.debug(f"EPGManager: Could not read EPG URL setting: {e}") + + return None + + def _get_default_time_range(self) -> tuple: + """ + Get default time range for EPG queries (now to +12 hours). + + Returns: + Tuple of (start_time, end_time) as Unix timestamps + """ + now = datetime.now() + start = int(now.timestamp()) + end = int((now + timedelta(hours=12)).timestamp()) + + logger.debug(f"EPGManager: Using default time range: {start} to {end}") + return start, end + + def get_epg( + self, + provider_name: str, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None + ) -> List[Dict]: + """ + Get EPG data for a specific channel within a time range. + + This is the main entry point for EPG queries. + + Args: + provider_name: Name of provider (e.g., "rtlplus", "joyn_de") + channel_id: Channel ID within provider + start_time: Start of time range (datetime), None for now + end_time: End of time range (datetime), None for now+12h + + Returns: + List of EPG entries as dictionaries (empty on any error) + """ + logger.info(f"EPGManager: Getting EPG for {provider_name}/{channel_id}") + + try: + # Step 1: Map to EPG channel ID + epg_channel_id = self.mapping.get_epg_channel_id(provider_name, channel_id) + if not epg_channel_id: + logger.warning( + f"EPGManager: No EPG mapping found for {provider_name}/{channel_id}" + ) + return [] + + logger.debug(f"EPGManager: Mapped to EPG channel ID: {epg_channel_id}") + + # Step 2: Get or download EPG XML + xml_path = self.cache.get_or_download(self.epg_url) + if not xml_path: + logger.error("EPGManager: Failed to get EPG XML file") + return [] + + # Step 3: Convert datetime to Unix timestamps + if start_time is None or end_time is None: + default_start, default_end = self._get_default_time_range() + start_ts = int(start_time.timestamp()) if start_time else default_start + end_ts = int(end_time.timestamp()) if end_time else default_end + else: + start_ts = int(start_time.timestamp()) + end_ts = int(end_time.timestamp()) + + logger.debug(f"EPGManager: Time range: {start_ts} to {end_ts}") + + # Step 4: Parse EPG for channel and time range + epg_entries = self.parser.parse_epg_for_channel( + xml_path, + epg_channel_id, + start_ts, + end_ts + ) + + logger.info( + f"EPGManager: Retrieved {len(epg_entries)} EPG entries for " + f"{provider_name}/{channel_id}" + ) + + return epg_entries + + except Exception as e: + logger.error(f"EPGManager: Error getting EPG: {e}", exc_info=True) + return [] + + def get_epg_for_provider( + self, + provider_name: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None + ) -> Dict[str, List[Dict]]: + """ + Get EPG data for all channels of a provider. + + Args: + provider_name: Name of provider + start_time: Start of time range (datetime) + end_time: End of time range (datetime) + + Returns: + Dictionary mapping channel IDs to their EPG entries + """ + logger.info(f"EPGManager: Getting EPG for all channels of provider '{provider_name}'") + + result = {} + + # Get all mapped channels for this provider + provider_mapping = self.mapping.get_provider_mapping(provider_name) + + if not provider_mapping: + logger.warning(f"EPGManager: No channels mapped for provider '{provider_name}'") + return result + + # Get EPG for each channel + for channel_id in provider_mapping.keys(): + epg_entries = self.get_epg(provider_name, channel_id, start_time, end_time) + if epg_entries: + result[channel_id] = epg_entries + + logger.info( + f"EPGManager: Retrieved EPG for {len(result)} channels of provider '{provider_name}'" + ) + + return result + + def clear_cache(self) -> bool: + """ + Clear EPG cache. + + Returns: + True if cleared successfully + """ + logger.info("EPGManager: Clearing EPG cache") + return self.cache.clear_cache() + + def reload_mapping(self) -> bool: + """ + Reload channel mapping from file. + + Returns: + True if reloaded successfully + """ + logger.info("EPGManager: Reloading channel mapping") + return self.mapping.reload_mapping() + + def get_cache_info(self) -> Optional[Dict]: + """ + Get information about EPG cache. + + Returns: + Dictionary with cache info, or None if no cache + """ + return self.cache.get_cache_info() + + def get_mapping_stats(self) -> Dict: + """ + Get statistics about channel mapping. + + Returns: + Dictionary with mapping statistics + """ + return self.mapping.get_mapping_stats() + + def has_mapping_for_channel(self, provider_name: str, channel_id: str) -> bool: + """ + Check if EPG mapping exists for a specific channel. + + Args: + provider_name: Name of provider + channel_id: Channel ID + + Returns: + True if mapping exists + """ + return self.mapping.has_mapping(provider_name, channel_id) \ No newline at end of file diff --git a/lib/streaming_providers/base/epg/epg_mapping.py b/lib/streaming_providers/base/epg/epg_mapping.py new file mode 100644 index 0000000..e69de29 diff --git a/lib/streaming_providers/base/epg/epg_parser.py b/lib/streaming_providers/base/epg/epg_parser.py new file mode 100644 index 0000000..28d6fdd --- /dev/null +++ b/lib/streaming_providers/base/epg/epg_parser.py @@ -0,0 +1,407 @@ +#!/usr/bin/env python3 +# streaming_providers/base/epg/epg_parser.py +""" +XMLTV EPG Parser for Kodi PVR Backend +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, Dict, Optional, Tuple +from ..utils.logger import logger + + +class EPGParser: + """ + Parses XMLTV EPG data and converts to Kodi PVR format. + Uses streaming parse for performance with large EPG files. + """ + + @staticmethod + def parse_xmltv_time(time_str: str) -> Optional[int]: + """ + Parse XMLTV time format to Unix timestamp. + Format: yyyyMMddHHmmss +HHMM + Example: 20240101200000 +0100 + + Args: + time_str: XMLTV formatted time string + + Returns: + Unix timestamp or None if parsing fails + """ + if not time_str: + return None + + try: + # Split timestamp and timezone + parts = time_str.strip().split() + if not parts: + return None + + dt_str = parts[0] + + # Parse basic timestamp + dt = datetime.strptime(dt_str, '%Y%m%d%H%M%S') + + # Handle timezone offset if present + if len(parts) > 1: + tz_str = parts[1] + if tz_str.startswith(('+', '-')): + sign = 1 if tz_str[0] == '+' else -1 + hours = int(tz_str[1:3]) + minutes = int(tz_str[3:5]) if len(tz_str) > 3 else 0 + # Adjust for timezone (subtract offset to get UTC) + from datetime import timedelta + dt = dt - timedelta(hours=sign * hours, minutes=sign * minutes) + + return int(dt.timestamp()) + + except (ValueError, IndexError) as e: + logger.warning(f"Failed to parse XMLTV time '{time_str}': {e}") + return None + + @staticmethod + def generate_broadcast_id(channel_id: str, start_time: int) -> int: + """ + Generate deterministic broadcast ID from channel and start time. + + Args: + channel_id: EPG channel ID + start_time: Unix timestamp of programme start + + 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) + + @staticmethod + def parse_episode_num(episode_elem: ET.Element) -> Tuple[Optional[int], Optional[int], Optional[int]]: + """ + Parse episode numbering from XMLTV episode-num element. + + Supports two systems: + - xmltv_ns: "season.episode.part/total" (0-indexed, e.g., "4.11.0/1" = S05E12) + - onscreen: "S05E12" format + + Args: + episode_elem: episode-num XML element + + Returns: + Tuple of (series_number, episode_number, episode_part_number) + """ + system = episode_elem.get('system', '') + text = episode_elem.text or '' + + if system == 'xmltv_ns': + # Format: "season.episode.part/total_parts" + # Example: "4.11.0/1" means season 5, episode 12, part 1 of 1 + try: + parts = text.split('.') + if len(parts) >= 2: + series_num = int(parts[0].split('/')[0]) + 1 if parts[0] else None # 0-indexed, so +1 + episode_num = int(parts[1].split('/')[0]) + 1 if parts[1] else None + part_num = None + if len(parts) >= 3 and parts[2]: + part_str = parts[2].split('/')[0] + part_num = int(part_str) + 1 if part_str else None + return series_num, episode_num, part_num + except (ValueError, IndexError) as e: + logger.debug(f"Failed to parse xmltv_ns episode number '{text}': {e}") + + elif system == 'onscreen': + # Format: "S05E12" or "5x12" + try: + import re + # Try S##E## format + match = re.match(r'S(\d+)E(\d+)', text, re.IGNORECASE) + if match: + return int(match.group(1)), int(match.group(2)), None + # Try #x## format + match = re.match(r'(\d+)x(\d+)', text) + if match: + return int(match.group(1)), int(match.group(2)), None + except (ValueError, AttributeError) as e: + logger.debug(f"Failed to parse onscreen episode number '{text}': {e}") + + return None, None, None + + @staticmethod + def parse_star_rating(rating_elem: ET.Element) -> Optional[int]: + """ + Parse star rating from XMLTV format to 0-10 scale. + + Args: + rating_elem: star-rating XML element + + Returns: + Rating as integer 0-10, or None + """ + value_elem = rating_elem.find('value') + if value_elem is not None and value_elem.text: + try: + # Format can be "8/10" or "4/5" or just "8" + text = value_elem.text.strip() + if '/' in text: + parts = text.split('/') + numerator = float(parts[0]) + denominator = float(parts[1]) + # Normalize to 0-10 scale + rating = int((numerator / denominator) * 10) + return min(10, max(0, rating)) + else: + # Assume already 0-10 scale + rating = int(float(text)) + return min(10, max(0, rating)) + except (ValueError, IndexError) as e: + logger.debug(f"Failed to parse star rating '{value_elem.text}': {e}") + return None + + @staticmethod + def parse_programme(programme_elem: ET.Element, epg_channel_id: str) -> Optional[Dict]: + """ + Parse a single XMLTV programme element to Kodi EPG format. + + Args: + programme_elem: XML programme element + epg_channel_id: EPG channel ID for validation + + Returns: + Dictionary with EPG data, or None if required fields missing + """ + # Verify channel matches + channel = programme_elem.get('channel', '') + if channel != epg_channel_id: + return None + + # Parse required fields + start_str = programme_elem.get('start', '') + stop_str = programme_elem.get('stop', '') + + start_time = EPGParser.parse_xmltv_time(start_str) + end_time = EPGParser.parse_xmltv_time(stop_str) + + if not start_time or not end_time: + logger.debug(f"Programme missing valid start/end time: start={start_str}, stop={stop_str}") + return None + + # Get title + 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) + + # Build base EPG entry with required fields + epg_entry = { + 'broadcast_id': broadcast_id, + 'title': title.strip(), + 'start_time': start_time, + 'end_time': end_time + } + + # Parse optional fields + + # Sub-title (episode name) + subtitle_elem = programme_elem.find('sub-title') + if subtitle_elem is not None and subtitle_elem.text: + epg_entry['episode_name'] = subtitle_elem.text.strip() + + # Description (plot) + desc_elem = programme_elem.find('desc') + if desc_elem is not None and desc_elem.text: + plot = desc_elem.text.strip() + epg_entry['plot'] = plot + # Use first sentence or first 100 chars as plot_outline + outline = plot.split('.')[0] if '.' in plot else plot[:100] + epg_entry['plot_outline'] = outline.strip() + + # Credits + credits_elem = programme_elem.find('credits') + if credits_elem is not None: + # Director + director_elem = credits_elem.find('director') + if director_elem is not None and director_elem.text: + epg_entry['director'] = director_elem.text.strip() + + # Actors (join with comma) + actors = credits_elem.findall('actor') + if actors: + cast_list = [actor.text.strip() for actor in actors if actor.text] + if cast_list: + epg_entry['cast'] = ', '.join(cast_list) + + # Writer + writer_elem = credits_elem.find('writer') + if writer_elem is not None and writer_elem.text: + epg_entry['writer'] = writer_elem.text.strip() + + # Year/Date + date_elem = programme_elem.find('date') + if date_elem is not None and date_elem.text: + try: + year = int(date_elem.text.strip()[:4]) # Take first 4 chars + epg_entry['year'] = year + except ValueError: + pass + + # Category (genre) + category_elem = programme_elem.find('category') + if category_elem is not None and category_elem.text: + epg_entry['genre_description'] = category_elem.text.strip() + + # Icon + icon_elem = programme_elem.find('icon') + if icon_elem is not None: + icon_src = icon_elem.get('src', '') + if icon_src: + epg_entry['icon_path'] = icon_src + + # Episode numbering (try all episode-num elements) + episode_nums = programme_elem.findall('episode-num') + series_num = episode_num = part_num = None + + # Try xmltv_ns first + for ep_elem in episode_nums: + if ep_elem.get('system') == 'xmltv_ns': + series_num, episode_num, part_num = EPGParser.parse_episode_num(ep_elem) + break + + # Fall back to onscreen if xmltv_ns didn't work + if series_num is None: + for ep_elem in episode_nums: + if ep_elem.get('system') == 'onscreen': + series_num, episode_num, part_num = EPGParser.parse_episode_num(ep_elem) + break + + if series_num is not None: + epg_entry['series_number'] = series_num + if episode_num is not None: + epg_entry['episode_number'] = episode_num + if part_num is not None: + epg_entry['episode_part_number'] = part_num + + # Star rating + star_rating_elem = programme_elem.find('star-rating') + if star_rating_elem is not None: + rating = EPGParser.parse_star_rating(star_rating_elem) + if rating is not None: + epg_entry['star_rating'] = rating + + # Parental rating + rating_elems = programme_elem.findall('rating') + for rating_elem in rating_elems: + value_elem = rating_elem.find('value') + if value_elem is not None and value_elem.text: + try: + # Try to extract numeric rating + import re + match = re.search(r'\d+', value_elem.text) + if match: + epg_entry['parental_rating'] = int(match.group()) + break + except (ValueError, AttributeError): + pass + + # Previously shown (first aired) + prev_shown_elem = programme_elem.find('previously-shown') + if prev_shown_elem is not None: + first_aired_str = prev_shown_elem.get('start', '') + if first_aired_str: + first_aired = EPGParser.parse_xmltv_time(first_aired_str) + if first_aired: + epg_entry['first_aired'] = first_aired + + return epg_entry + + @staticmethod + def open_xml_file(file_path: str): + """ + Open XML file, handling both plain and gzipped files. + + Args: + file_path: Path to XML file (can be .xml or .xml.gz) + + Returns: + File object suitable for ET.iterparse + """ + if file_path.endswith('.gz'): + logger.debug(f"Opening gzipped EPG file: {file_path}") + return gzip.open(file_path, 'rt', encoding='utf-8') + else: + logger.debug(f"Opening plain EPG file: {file_path}") + return open(file_path, 'r', encoding='utf-8') + + @staticmethod + def parse_epg_for_channel( + xml_path: str, + epg_channel_id: str, + start_time: Optional[int] = None, + end_time: Optional[int] = None + ) -> List[Dict]: + """ + Parse EPG data for a specific channel within a time range. + Uses streaming parse for performance. + + Args: + xml_path: Path to XMLTV file (can be .xml or .xml.gz) + 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 + + Returns: + List of EPG entries as dictionaries + """ + logger.info(f"Parsing EPG for channel '{epg_channel_id}' from {xml_path}") + + programmes = [] + + try: + with EPGParser.open_xml_file(xml_path) as xml_file: + # Use iterparse for memory-efficient streaming + context = ET.iterparse(xml_file, events=('end',)) + + for event, elem in context: + 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) + + if epg_entry: + # Filter by time range + prog_start = epg_entry['start_time'] + prog_end = epg_entry['end_time'] + + # Include if programme overlaps with requested range + if start_time is not None and prog_end < start_time: + # Programme ends before requested range + elem.clear() + continue + + if end_time is not None and prog_start > end_time: + # Programme starts after requested range + # Since XMLTV is chronological, we can stop here + elem.clear() + break + + programmes.append(epg_entry) + + # Clear element to free memory + elem.clear() + + logger.info(f"Parsed {len(programmes)} programmes for channel '{epg_channel_id}'") + return programmes + + except ET.ParseError as e: + logger.error(f"XML parse error in EPG file: {e}") + return [] + except Exception as e: + logger.error(f"Error parsing EPG file: {e}") + return [] \ No newline at end of file diff --git a/lib/streaming_providers/base/models/streaming_channel.py b/lib/streaming_providers/base/models/streaming_channel.py index 3d45df6..599e82b 100644 --- a/lib/streaming_providers/base/models/streaming_channel.py +++ b/lib/streaming_providers/base/models/streaming_channel.py @@ -1,12 +1,47 @@ # streaming_providers/base/models.py from dataclasses import dataclass -from typing import Dict, Optional +from enum import Enum +from typing import Dict, Optional, List +import logging from .drm_models import DRMConfig +# Setup logging for validation warnings +logger = logging.getLogger(__name__) + + +class StreamingMode(Enum): + """Enum for streaming modes""" + LIVE = "live" + VOD = "vod" + + +class ContentType(Enum): + """Enum for content types""" + LIVE = "LIVE" + VOD = "VOD" + SERIES = "SERIES" + MOVIE = "MOVIE" + RADIO = "RADIO" + + +class Quality(Enum): + """Enum for stream quality""" + SD = "SD" + HD = "HD" + UHD = "UHD" + FOUR_K = "4K" + AUDIO = "AUDIO" + + @dataclass class StreamingChannel: """ Universal channel representation for all providers + + Backward Compatibility Note: + - All existing fields remain unchanged + - New fields have default values + - All existing methods remain functional """ # Core identification name: str @@ -21,7 +56,7 @@ class StreamingChannel: quality: Optional[str] = None # 'SD', 'HD', 'UHD', '4K' # Streaming configuration - mode: str = "live" # "live" or "vod" + mode: str = "live" # "live" or "vod" - kept as string for compatibility session_manifest: bool = False manifest: Optional[str] = None manifest_script: Optional[str] = None @@ -52,8 +87,62 @@ class StreamingChannel: certificate_url: Optional[str] = None streaming_format: Optional[str] = None + # NEW: Radio support (backward compatible - defaults to False) + is_radio: bool = False + audio_only: bool = False + radio_frequency: Optional[str] = None # e.g., "FM 98.5" + radio_station_id: Optional[str] = None + bitrate: Optional[str] = None # e.g., "128kbps" + + def __post_init__(self): + """ + Post-initialization processing for backward compatibility + and automatic field synchronization + """ + # Run validation checks (warnings only for backward compatibility) + self._validate_fields() + + # Auto-set audio_only if is_radio is True + if self.is_radio: + self.audio_only = True + + # Auto-update content_type for radio + if self.is_radio and self.content_type == 'LIVE': + self.content_type = 'RADIO' + + # Ensure quality is appropriate for audio content + if self.audio_only and not self.quality: + self.quality = 'AUDIO' + + def _validate_fields(self): + """ + Internal validation method that logs warnings instead of raising errors + for backward compatibility + """ + # Validate content_type consistency + if self.mode == 'vod' and self.content_type == 'LIVE': + logger.warning( + f"Channel {self.name} ({self.channel_id}): " + f"VOD mode with LIVE content_type - consider changing to VOD" + ) + + # Validate manifest consistency + if self.session_manifest and self.manifest: + logger.warning( + f"Channel {self.name} ({self.channel_id}): " + f"session_manifest=True but manifest URL is set - manifest will be ignored" + ) + + # Validate radio fields + if self.is_radio and not self.audio_only: + logger.info( + f"Channel {self.name} ({self.channel_id}): " + f"is_radio=True but audio_only=False - auto-setting audio_only=True" + ) + self.audio_only = True + def to_dict(self) -> Dict: - """Convert to dictionary format""" + """Convert to dictionary format - backward compatible with new fields added""" result = { 'Name': self.name, 'Id': self.channel_id, @@ -75,7 +164,13 @@ class StreamingChannel: 'ContentType': self.content_type, 'Country': self.country, 'Language': self.language, - 'StreamingFormat': self.streaming_format + 'StreamingFormat': self.streaming_format, + # NEW: Radio fields (backward compatible addition) + 'IsRadio': self.is_radio, + 'AudioOnly': self.audio_only, + 'RadioFrequency': self.radio_frequency, + 'RadioStationId': self.radio_station_id, + 'Bitrate': self.bitrate } # Add DRM config if present @@ -102,3 +197,149 @@ class StreamingChannel: self.manifest = None self.session_manifest = True self.manifest_script = manifest_script_params + + # NEW: Backward compatible enhancements + + @classmethod + def create_live_channel(cls, name: str, channel_id: str, provider: str, **kwargs) -> 'StreamingChannel': + """ + Factory method for live channels with proper defaults + Backward compatible: Existing code can continue using direct instantiation + """ + return cls( + name=name, + channel_id=channel_id, + provider=provider, + mode="live", + content_type="LIVE", + **kwargs + ) + + @classmethod + def create_vod_channel(cls, name: str, content_id: str, provider: str, **kwargs) -> 'StreamingChannel': + """ + Factory method for VOD content + Backward compatible: Existing code can continue using direct instantiation + """ + return cls( + name=name, + channel_id=content_id, + provider=provider, + mode="vod", + content_type="VOD", + **kwargs + ) + + @classmethod + def create_radio_channel(cls, name: str, channel_id: str, provider: str, **kwargs) -> 'StreamingChannel': + """ + Factory method for radio channels + Backward compatible: Existing code can continue using direct instantiation + """ + return cls( + name=name, + channel_id=channel_id, + provider=provider, + is_radio=True, + audio_only=True, + content_type="RADIO", + quality="AUDIO", + **kwargs + ) + + def get_streaming_urls(self) -> List[str]: + """ + Extract all relevant URLs for logging/validation + New method - doesn't affect backward compatibility + """ + urls = [] + if self.manifest: + urls.append(self.manifest) + if self.license_url: + urls.append(self.license_url) + if self.certificate_url: + urls.append(self.certificate_url) + return urls + + def requires_drm(self) -> bool: + """ + Check if this channel needs DRM handling + New method - doesn't affect backward compatibility + """ + return bool(self.drm_config) or bool(self.license_url) + + def is_audio_content(self) -> bool: + """ + Check if this is audio-only content (radio or audio track) + New method - doesn't affect backward compatibility + """ + return self.is_radio or self.audio_only + + def detect_and_set_radio(self) -> None: + """ + Auto-detect if this is likely a radio channel + New method - doesn't affect backward compatibility + """ + if self.is_radio: # Already set + return + + radio_indicators = [ + self.name.lower().startswith('radio'), + 'radio' in self.name.lower(), + self.quality in ['audio', 'aac', 'mp3', 'AUDIO'], + self.description and 'radio' in self.description.lower(), + self.genre and 'radio' in self.genre.lower(), + ] + + if any(radio_indicators): + self.is_radio = True + self.audio_only = True + + # Update quality if not set + if not self.quality or self.quality.upper() not in [q.value for q in Quality]: + self.quality = 'AUDIO' + + # Update content_type if it's still LIVE + if self.content_type == 'LIVE': + self.content_type = 'RADIO' + + # Compatibility properties (optional - for clearer naming) + @property + def dynamic_manifest(self) -> bool: + """Alias for session_manifest with clearer name""" + return self.session_manifest + + @dynamic_manifest.setter + def dynamic_manifest(self, value: bool): + """Setter for dynamic_manifest alias""" + self.session_manifest = value + + @property + def requires_session_manifest(self) -> bool: + """Alternative property name for clarity""" + return self.session_manifest + + def validate(self) -> List[str]: + """ + Run comprehensive validation and return list of warnings/issues + New method - doesn't affect backward compatibility + """ + warnings = [] + + # Check for missing required fields for streaming + if not self.manifest and not self.manifest_script: + warnings.append("No manifest URL or manifest script provided") + + # Check DRM configuration + if self.license_url and not self.drm_config: + warnings.append("License URL provided but no DRM configuration") + + # Check content type consistency + if self.mode == 'vod' and self.content_type == 'LIVE': + warnings.append("VOD mode should not have LIVE content_type") + + # Check radio consistency + if self.is_radio and self.quality not in ['AUDIO', 'audio', None]: + warnings.append(f"Radio channel has video quality setting: {self.quality}") + + return warnings \ No newline at end of file diff --git a/resources/config/epg_mapping.json b/resources/config/epg_mapping.json new file mode 100644 index 0000000..3409c69 --- /dev/null +++ b/resources/config/epg_mapping.json @@ -0,0 +1,42 @@ +{ + "_comment": "EPG Channel Mapping - Maps provider/channel IDs to XMLTV EPG channel IDs", + "_format": "{ 'provider_name': { 'channel_id': 'epg_channel_id' } }", + "_note": "Provider names include country suffix for multi-country providers (e.g., joyn_de, joyn_ch)", + + "rtlplus": { + "rtl": "de.rtl", + "vox": "de.vox", + "rtl2": "de.rtl2", + "nitro": "de.nitro", + "super_rtl": "de.superrtl" + }, + + "joyn_de": { + "prosieben": "de.prosieben", + "sat1": "de.sat1", + "kabel_eins": "de.kabeleins", + "sixx": "de.sixx", + "pro7_maxx": "de.pro7maxx", + "sat1_gold": "de.sat1gold" + }, + + "joyn_ch": { + "prosieben_schweiz": "ch.prosieben", + "sat1_schweiz": "ch.sat1", + "kabel_eins_schweiz": "ch.kabeleins" + }, + + "zattoo_de": { + "ard": "de.ard", + "zdf": "de.zdf", + "rtl": "de.rtl", + "sat1": "de.sat1", + "prosieben": "de.prosieben" + }, + + "example_provider": { + "channel1": "epg.channel.one", + "channel2": "epg.channel.two", + "channel3": "epg.channel.three" + } +} \ No newline at end of file