From 75c2f5d95cb463718b9f158623fe5360ab07bfba Mon Sep 17 00:00:00 2001 From: Nirvana Date: Thu, 4 Dec 2025 10:22:54 +0100 Subject: [PATCH] Implement epg --- lib/streaming_providers/base/epg/epg_cache.py | 252 ++++++++++++++++++ .../base/epg/epg_manager.py | 62 ++--- .../base/epg/epg_mapping.py | 240 +++++++++++++++++ lib/streaming_providers/base/manager.py | 88 +++++- resources/settings.xml | 14 +- service.py | 2 +- 6 files changed, 604 insertions(+), 54 deletions(-) diff --git a/lib/streaming_providers/base/epg/epg_cache.py b/lib/streaming_providers/base/epg/epg_cache.py index e69de29..b095afe 100644 --- a/lib/streaming_providers/base/epg/epg_cache.py +++ b/lib/streaming_providers/base/epg/epg_cache.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +# streaming_providers/base/epg/epg_cache.py +""" +EPG Cache Manager using VFS +Handles downloading, caching, and TTL management for EPG XML files +""" + +import time +from typing import Optional, Dict +from ..utils.logger import logger +from ..utils.vfs import VFS + + +class EPGCache: + """ + Manages EPG XML file caching with VFS backend. + Handles both plain and gzipped XML files. + """ + + # Cache TTL: 24 hours + CACHE_TTL_SECONDS = 24 * 60 * 60 + + # File names + EPG_FILE = "epg.xml" + EPG_GZ_FILE = "epg.xml.gz" + METADATA_FILE = "epg_metadata.json" + + def __init__(self, vfs_subdir: str = "epg_cache"): + """ + Initialize EPG cache manager. + + Args: + vfs_subdir: Subdirectory under addon data for EPG cache + """ + self.vfs = VFS(addon_subdir=vfs_subdir) + logger.info(f"EPGCache: Initialized with VFS path: {self.vfs.base_path}") + + def _get_metadata(self) -> Optional[Dict]: + """ + Load cache metadata from VFS. + + Returns: + Metadata dictionary or None if not found/invalid + """ + metadata = self.vfs.read_json(self.METADATA_FILE) + if metadata: + logger.debug(f"EPGCache: Loaded metadata: {metadata}") + return metadata + + def _save_metadata(self, url: str, file_size: int, is_gzipped: bool) -> bool: + """ + Save cache metadata to VFS. + + Args: + url: EPG source URL + file_size: Size of cached file in bytes + is_gzipped: Whether file is gzipped + + Returns: + True if saved successfully + """ + metadata = { + 'downloaded_at': int(time.time()), + 'url': url, + 'file_size': file_size, + 'is_gzipped': is_gzipped + } + + success = self.vfs.write_json(self.METADATA_FILE, metadata) + if success: + logger.info(f"EPGCache: Saved metadata for {url}") + else: + logger.error(f"EPGCache: Failed to save metadata") + return success + + def is_cache_valid(self) -> bool: + """ + Check if cached EPG is still valid (within TTL). + + Returns: + True if cache exists and is valid + """ + metadata = self._get_metadata() + if not metadata: + logger.debug("EPGCache: No metadata found, cache invalid") + return False + + # Check if EPG file exists + filename = self.EPG_GZ_FILE if metadata.get('is_gzipped') else self.EPG_FILE + if not self.vfs.exists(filename): + logger.debug(f"EPGCache: EPG file '{filename}' not found, cache invalid") + return False + + # Check TTL + downloaded_at = metadata.get('downloaded_at', 0) + age = int(time.time()) - downloaded_at + + if age > self.CACHE_TTL_SECONDS: + logger.info(f"EPGCache: Cache expired (age: {age}s, TTL: {self.CACHE_TTL_SECONDS}s)") + return False + + logger.debug(f"EPGCache: Cache valid (age: {age}s)") + return True + + def get_cached_file_path(self) -> Optional[str]: + """ + Get path to cached EPG file if valid. + + Returns: + Full path to EPG file, or None if cache invalid + """ + if not self.is_cache_valid(): + return None + + metadata = self._get_metadata() + if not metadata: + return None + + filename = self.EPG_GZ_FILE if metadata.get('is_gzipped') else self.EPG_FILE + file_path = self.vfs.join_path(filename) + + logger.debug(f"EPGCache: Returning cached file path: {file_path}") + return file_path + + def download_and_cache(self, url: str) -> Optional[str]: + """ + Download EPG file from URL and cache it. + Handles both plain and gzipped files automatically. + + Args: + url: URL to download EPG from + + Returns: + Path to cached file, or None on failure + """ + logger.info(f"EPGCache: Downloading EPG from {url}") + + try: + import requests + + # Download with streaming to handle large files + response = requests.get(url, stream=True, timeout=60) + response.raise_for_status() + + # Determine if content is gzipped + content_type = response.headers.get('Content-Type', '').lower() + content_encoding = response.headers.get('Content-Encoding', '').lower() + is_gzipped = ( + 'gzip' in content_encoding or + url.endswith('.gz') or + 'gzip' in content_type + ) + + filename = self.EPG_GZ_FILE if is_gzipped else self.EPG_FILE + + # Get full file path + file_path = self.vfs.join_path(filename) + + # Ensure directory exists + self.vfs.ensure_directory(file_path) + + # Download in chunks + chunk_size = 8192 + total_size = 0 + + with open(file_path, 'wb') as f: + for chunk in response.iter_content(chunk_size=chunk_size): + if chunk: + f.write(chunk) + total_size += len(chunk) + + logger.info(f"EPGCache: Downloaded {total_size} bytes to {filename}") + + # Save metadata + self._save_metadata(url, total_size, is_gzipped) + + return file_path + + except Exception as e: + logger.error(f"EPGCache: Download failed: {e}", exc_info=True) + return None + + def get_or_download(self, url: str) -> Optional[str]: + """ + Get cached EPG file path, or download if cache is invalid. + + Args: + url: URL to download from if cache invalid + + Returns: + Path to EPG file (cached or freshly downloaded), or None on failure + """ + # Try cache first + cached_path = self.get_cached_file_path() + if cached_path: + logger.info("EPGCache: Using cached EPG file") + return cached_path + + # Cache miss or expired - download new + logger.info("EPGCache: Cache miss or expired, downloading") + return self.download_and_cache(url) + + def clear_cache(self) -> bool: + """ + Clear all cached EPG files and metadata. + + Returns: + True if cleared successfully + """ + logger.info("EPGCache: Clearing cache") + + success = True + + # Delete EPG files + for filename in [self.EPG_FILE, self.EPG_GZ_FILE, self.METADATA_FILE]: + if self.vfs.exists(filename): + if not self.vfs.delete(filename): + logger.warning(f"EPGCache: Failed to delete {filename}") + success = False + + if success: + logger.info("EPGCache: Cache cleared successfully") + + return success + + def get_cache_info(self) -> Optional[Dict]: + """ + Get information about cached EPG. + + Returns: + Dictionary with cache info, or None if no cache + """ + metadata = self._get_metadata() + if not metadata: + return None + + filename = self.EPG_GZ_FILE if metadata.get('is_gzipped') else self.EPG_FILE + file_exists = self.vfs.exists(filename) + + age = int(time.time()) - metadata.get('downloaded_at', 0) + is_valid = self.is_cache_valid() + + return { + 'url': metadata.get('url'), + 'downloaded_at': metadata.get('downloaded_at'), + 'age_seconds': age, + 'file_size': metadata.get('file_size'), + 'is_gzipped': metadata.get('is_gzipped'), + 'file_exists': file_exists, + 'is_valid': is_valid, + 'ttl_seconds': self.CACHE_TTL_SECONDS + } \ No newline at end of file diff --git a/lib/streaming_providers/base/epg/epg_manager.py b/lib/streaming_providers/base/epg/epg_manager.py index 72300c5..63c6a06 100644 --- a/lib/streaming_providers/base/epg/epg_manager.py +++ b/lib/streaming_providers/base/epg/epg_manager.py @@ -17,32 +17,24 @@ class EPGManager: """ Central manager for EPG operations. Coordinates cache management, channel mapping, and EPG parsing. + + All file operations are handled transparently by VFS - works in both + Kodi and standard Python environments. """ # 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 - ): + def __init__(self, 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 + # Initialize components - they handle paths internally self.cache = EPGCache(vfs_subdir="epg_cache") - self.mapping = EPGMapping(addon_path, user_data_path) + self.mapping = EPGMapping() # No paths needed - handles internally self.parser = EPGParser() # Get EPG URL from settings or use default @@ -50,48 +42,24 @@ class EPGManager: 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. + Get EPG URL from addon settings using SettingsManager. Returns: URL from settings, or None if not set """ try: - import xbmcaddon - addon = xbmcaddon.Addon() - url = addon.getSetting('epg_xml_url') + from ..settings.kodi_settings_bridge import KodiSettingsBridge - if url and url.strip(): - logger.info(f"EPGManager: Using EPG URL from settings: {url}") - return url.strip() + bridge = KodiSettingsBridge() + if bridge.is_kodi_environment(): + url = bridge.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}") diff --git a/lib/streaming_providers/base/epg/epg_mapping.py b/lib/streaming_providers/base/epg/epg_mapping.py index e69de29..b71a9ad 100644 --- a/lib/streaming_providers/base/epg/epg_mapping.py +++ b/lib/streaming_providers/base/epg/epg_mapping.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +# streaming_providers/base/epg/epg_mapping.py +""" +EPG Channel Mapping Manager +Maps provider/channel IDs to EPG channel IDs from XMLTV +""" + +import os +from typing import Optional, Dict +from ..utils.logger import logger +from ..utils.vfs import VFS + + +class EPGMapping: + """ + Manages mapping between provider channel IDs and EPG channel IDs. + Supports user-editable mapping with default fallback. + + Uses VFS for all file operations. Automatically detects addon resources + path for default mapping (only used on first run to copy defaults). + """ + + # Mapping filename + MAPPING_FILE = "epg_mapping.json" + + def __init__(self): + """ + Initialize EPG mapping manager. + Automatically detects environment and sets up paths. + """ + # VFS for user data directory (where editable mapping lives) + self.user_vfs = VFS(addon_subdir="") # Root of addon data + + self.mapping: Dict[str, Dict[str, str]] = {} + + logger.info(f"EPGMapping: User mapping path: {self.user_vfs.base_path}") + + # Load mapping on initialization + self._load_mapping() + + def _get_default_mapping_path(self) -> Optional[str]: + """ + Get path to default mapping in addon resources. + Only used on first run to copy defaults to user directory. + + Uses settings manager to detect paths if available. + + Returns: + Path to default mapping file, or None if not found + """ + try: + # Try to use settings manager's kodi bridge to get addon path + from ..settings.kodi_settings_bridge import KodiSettingsBridge + + bridge = KodiSettingsBridge() + if bridge.is_kodi_environment(): + addon_info = bridge.get_addon_info() + addon_path = addon_info.get('path') + if addon_path: + default_path = os.path.join(addon_path, 'resources', 'config', self.MAPPING_FILE) + logger.debug(f"EPGMapping: Default mapping path (Kodi): {default_path}") + return default_path + + # Fallback to standard filesystem + addon_path = os.getcwd() + default_path = os.path.join(addon_path, 'resources', 'config', self.MAPPING_FILE) + logger.debug(f"EPGMapping: Default mapping path (standard): {default_path}") + return default_path + + except Exception as e: + logger.warning(f"EPGMapping: Could not determine default mapping path: {e}") + # Last resort fallback + addon_path = os.getcwd() + default_path = os.path.join(addon_path, 'resources', 'config', self.MAPPING_FILE) + return default_path + + def _copy_default_to_user(self) -> bool: + """ + Copy default mapping from addon resources to user directory. + Only called on first run when user mapping doesn't exist. + + Returns: + True if copied successfully or created empty mapping + """ + default_path = self._get_default_mapping_path() + + if default_path and os.path.exists(default_path): + # Read default mapping using standard file operations + # (it's in addon resources, not in VFS user data) + try: + import json + with open(default_path, 'r', encoding='utf-8') as f: + default_mapping = json.load(f) + + # Write to user VFS + if self.user_vfs.write_json(self.MAPPING_FILE, default_mapping): + logger.info(f"EPGMapping: Copied default mapping to user directory") + return True + else: + logger.error(f"EPGMapping: Failed to write default mapping to user directory") + return False + + except Exception as e: + logger.error(f"EPGMapping: Failed to read/copy default mapping: {e}") + # Fall through to create empty mapping + + # No default found or copy failed - create empty mapping + logger.warning("EPGMapping: No default mapping found, creating empty mapping") + empty_mapping = {} + if self.user_vfs.write_json(self.MAPPING_FILE, empty_mapping): + logger.info("EPGMapping: Created empty user mapping file") + return True + else: + logger.error("EPGMapping: Failed to create empty user mapping") + return False + + def _ensure_user_mapping_exists(self) -> bool: + """ + Ensure user mapping file exists. Copy from default if needed. + + Returns: + True if user mapping exists or was created successfully + """ + # Check if user mapping already exists + if self.user_vfs.exists(self.MAPPING_FILE): + logger.debug("EPGMapping: User mapping file exists") + return True + + # User mapping doesn't exist - copy from default or create empty + logger.info("EPGMapping: User mapping not found, initializing...") + return self._copy_default_to_user() + + def _load_mapping(self) -> bool: + """ + Load mapping from user file. + + Returns: + True if mapping loaded successfully + """ + # Ensure user mapping exists (copy from default on first run) + self._ensure_user_mapping_exists() + + # Load user mapping + user_mapping = self.user_vfs.read_json(self.MAPPING_FILE) + if user_mapping is not None: + self.mapping = user_mapping + logger.info(f"EPGMapping: Loaded mapping with {len(self.mapping)} providers") + return True + + # Failed to load mapping + logger.warning("EPGMapping: Failed to load mapping, using empty mapping") + self.mapping = {} + return False + + def get_epg_channel_id(self, provider_name: str, channel_id: str) -> Optional[str]: + """ + Get EPG channel ID for a provider/channel combination. + + Args: + provider_name: Name of provider (e.g., "rtlplus", "joyn_de") + channel_id: Channel ID within provider + + Returns: + EPG channel ID (e.g., "de.rtl"), or None if not mapped + """ + # Get provider's mapping + provider_mapping = self.mapping.get(provider_name) + if not provider_mapping: + logger.debug(f"EPGMapping: No mapping found for provider '{provider_name}'") + return None + + # Get channel's EPG ID + epg_id = provider_mapping.get(channel_id) + if not epg_id: + logger.debug(f"EPGMapping: No EPG ID found for '{provider_name}/{channel_id}'") + return None + + logger.debug(f"EPGMapping: Mapped '{provider_name}/{channel_id}' -> '{epg_id}'") + return epg_id + + def get_provider_mapping(self, provider_name: str) -> Dict[str, str]: + """ + Get all channel mappings for a specific provider. + + Args: + provider_name: Name of provider + + Returns: + Dictionary mapping channel IDs to EPG IDs + """ + return self.mapping.get(provider_name, {}) + + def has_mapping(self, provider_name: str, channel_id: Optional[str] = None) -> bool: + """ + Check if mapping exists for provider or specific channel. + + Args: + provider_name: Name of provider + channel_id: Optional channel ID to check + + Returns: + True if mapping exists + """ + if provider_name not in self.mapping: + return False + + if channel_id is None: + return True + + return channel_id in self.mapping[provider_name] + + def reload_mapping(self) -> bool: + """ + Reload mapping from file (useful after user edits). + + Returns: + True if reloaded successfully + """ + logger.info("EPGMapping: Reloading mapping from file") + return self._load_mapping() + + def get_mapping_stats(self) -> Dict: + """ + Get statistics about current mapping. + + Returns: + Dictionary with mapping statistics + """ + total_providers = len(self.mapping) + total_channels = sum(len(channels) for channels in self.mapping.values()) + + user_mapping_path = self.user_vfs.join_path(self.MAPPING_FILE) + + return { + 'total_providers': total_providers, + 'total_channels': total_channels, + 'providers': list(self.mapping.keys()), + 'user_mapping_path': user_mapping_path, + 'user_mapping_exists': self.user_vfs.exists(self.MAPPING_FILE) + } \ No newline at end of file diff --git a/lib/streaming_providers/base/manager.py b/lib/streaming_providers/base/manager.py index 3fa985b..2681170 100644 --- a/lib/streaming_providers/base/manager.py +++ b/lib/streaming_providers/base/manager.py @@ -3,6 +3,7 @@ from typing import Dict, List, Optional from .provider import StreamingProvider from .models import StreamingChannel, DRMSystem from .drm import DRMPluginManager +from .epg import EPGManager from .utils.logger import logger class ProviderManager: @@ -17,6 +18,10 @@ class ProviderManager: self.drm_plugin_manager = DRMPluginManager() logger.info("ProviderManager: Initialized with DRM plugin manager") + self.epg_manager = EPGManager() + + logger.info("ProviderManager: Initialized with DRM plugin manager and EPG manager") + @staticmethod def _is_provider_enabled(provider_name: str, country: Optional[str] = None) -> bool: """ @@ -338,8 +343,23 @@ class ProviderManager: logger.error(f"ProviderManager: Cannot get EPG - provider '{provider_name}' not found") raise ValueError(f"Provider '{provider_name}' not found") - epg_data = provider.get_epg(channel_id, **kwargs) - logger.debug(f"ProviderManager: Retrieved {len(epg_data)} EPG entries for channel '{channel_id}' from provider '{provider_name}'") + # CHECK IF PROVIDER IMPLEMENTS ITS OWN EPG + if provider.implements_epg: + # Use provider's native EPG implementation + logger.debug(f"ProviderManager: Using native EPG for provider '{provider_name}'") + epg_data = provider.get_epg(channel_id, **kwargs) + else: + # Use generic EPG manager + logger.debug(f"ProviderManager: Using generic EPG for provider '{provider_name}'") + epg_data = self.epg_manager.get_epg( + provider_name=provider_name, + channel_id=channel_id, + start_time=kwargs.get('start_time'), + end_time=kwargs.get('end_time') + ) + + logger.debug( + f"ProviderManager: Retrieved {len(epg_data)} EPG entries for channel '{channel_id}' from provider '{provider_name}'") return epg_data def get_provider_epg_xmltv(self, provider_name: str, **kwargs) -> Optional[str]: @@ -361,13 +381,75 @@ class ProviderManager: logger.error(f"ProviderManager: Cannot get XMLTV EPG - provider '{provider_name}' not found") raise ValueError(f"Provider '{provider_name}' not found") - xmltv_data = provider.get_epg_xmltv(**kwargs) + # Only use provider's XMLTV if it implements EPG + if provider.implements_epg: + logger.debug(f"ProviderManager: Using native XMLTV EPG for provider '{provider_name}'") + xmltv_data = provider.get_epg_xmltv(**kwargs) + else: + # For generic EPG, we don't generate XMLTV format + # (The EPG XML is already in XMLTV format, but it's global, not provider-specific) + logger.warning( + f"ProviderManager: Provider '{provider_name}' does not implement EPG, no provider-specific XMLTV available") + xmltv_data = None + if xmltv_data: logger.info(f"ProviderManager: Retrieved XMLTV EPG data for provider '{provider_name}'") else: logger.warning(f"ProviderManager: No XMLTV EPG data available for provider '{provider_name}'") + return xmltv_data + def clear_epg_cache(self) -> bool: + """ + Clear the generic EPG cache. + + Returns: + True if cleared successfully + """ + logger.info("ProviderManager: Clearing EPG cache") + return self.epg_manager.clear_cache() + + def reload_epg_mapping(self) -> bool: + """ + Reload EPG channel mapping from file. + + Returns: + True if reloaded successfully + """ + logger.info("ProviderManager: Reloading EPG mapping") + return self.epg_manager.reload_mapping() + + def get_epg_cache_info(self) -> Optional[Dict]: + """ + Get information about EPG cache. + + Returns: + Dictionary with cache info, or None if no cache + """ + return self.epg_manager.get_cache_info() + + def get_epg_mapping_stats(self) -> Dict: + """ + Get statistics about EPG channel mapping. + + Returns: + Dictionary with mapping statistics + """ + return self.epg_manager.get_mapping_stats() + + def has_epg_mapping(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.epg_manager.has_mapping_for_channel(provider_name, channel_id) + def get_channel_drm_configs(self, provider_name: str, channel_id: str, **kwargs) -> List: provider = self.get_provider(provider_name) if not provider: diff --git a/resources/settings.xml b/resources/settings.xml index 663e423..ee2b35c 100644 --- a/resources/settings.xml +++ b/resources/settings.xml @@ -2,10 +2,18 @@ + + 1 + URL to download XMLTV EPG data. Can be .xml or .xml.gz format. Leave empty to use default. + + + + + 2 + Force clear EPG cache and re-download on next request + RunScript(special://home/addons/YOUR_ADDON_ID/clear_epg_cache.py) + - - - diff --git a/service.py b/service.py index cfd625c..721b170 100644 --- a/service.py +++ b/service.py @@ -209,7 +209,7 @@ class UltimateService: provider_label = provider_name # Add M3U entry with extended info first - entry_content += f'#EXTINF:-1 tvg-logo="{channel_logo}" group-title="{provider_label}",{channel_name}\n' + entry_content += f'#EXTINF:-1 tvg-id="{channel_id}" tvg-logo="{channel_logo}" group-title="{provider_label}",{channel_name}\n' # Get DRM configs and add KODIPROP directives try: