From f504776967280dd490861272a6041f07b89a807c Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 12 Dec 2025 12:03:13 +0100 Subject: [PATCH] Add web config --- docker-compose.yml | 1 - .../base/settings/provider_enable_manager.py | 496 ++++++++++++++++++ .../base/settings/settings_manager.py | 43 +- service.py | 221 ++++++-- 4 files changed, 691 insertions(+), 70 deletions(-) create mode 100644 lib/streaming_providers/base/settings/provider_enable_manager.py diff --git a/docker-compose.yml b/docker-compose.yml index adcd663..c97a401 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -14,7 +14,6 @@ services: - "7777:7777" environment: - ULTIMATE_PORT=7777 - - ULTIMATE_COUNTRY=DE - ULTIMATE_DEBUG=false - TZ=${TZ:-Europe/Berlin} volumes: diff --git a/lib/streaming_providers/base/settings/provider_enable_manager.py b/lib/streaming_providers/base/settings/provider_enable_manager.py new file mode 100644 index 0000000..dccc00f --- /dev/null +++ b/lib/streaming_providers/base/settings/provider_enable_manager.py @@ -0,0 +1,496 @@ +# streaming_providers/base/settings/provider_enable_manager.py +""" +Provider Enable Manager - Handles provider enable/disable status with precedence: +1. Kodi settings (if in Kodi environment and setting exists) +2. providers_enabled.json file +3. Default to enabled (True) + +Works in both Kodi and standalone environments. +""" + +import time +from typing import Dict, Optional, Tuple, Any +from enum import Enum + +from ..utils.logger import logger +from ..utils.environment import is_kodi_environment, get_vfs_instance + + +class EnableSource(Enum): + """Source of the enable setting""" + KODI = "kodi" + FILE = "file" + DEFAULT = "default" + + +class ProviderEnableManager: + """ + Manages provider enable/disable status with proper precedence. + Uses VFS for file operations to work in both Kodi and standalone. + """ + + DEFAULT_FILENAME = "providers_enabled.json" + DEFAULT_VERSION = "1.0" + CACHE_TTL = 30 # seconds + + def __init__(self, config_dir: Optional[str] = None): + """ + Initialize the enable manager. + + Args: + config_dir: Optional configuration directory (uses VFS default if None) + """ + self.vfs = get_vfs_instance(config_dir=config_dir) + self._cache: Optional[Dict[str, Any]] = None + self._cache_time: float = 0 + self._kodi_bridge = None + + # Try to initialize Kodi bridge if in Kodi environment + if is_kodi_environment(): + try: + from .settings.kodi_settings_bridge import KodiSettingsBridge + self._kodi_bridge = KodiSettingsBridge(config_dir=config_dir) + logger.debug("ProviderEnableManager: Kodi bridge initialized") + except ImportError as e: + logger.debug(f"ProviderEnableManager: Kodi bridge not available: {e}") + self._kodi_bridge = None + + def _load_file(self, force_reload: bool = False) -> Dict[str, Any]: + """ + Load providers_enabled.json file with caching. + + Args: + force_reload: Force reload from disk ignoring cache + + Returns: + Dictionary with file contents + """ + current_time = time.time() + + # Check cache first + if not force_reload and self._cache and (current_time - self._cache_time) < self.CACHE_TTL: + return self._cache + + # Initialize default structure + default_data = { + "version": self.DEFAULT_VERSION, + "providers": {} + } + + # Check if file exists + if not self.vfs.exists(self.DEFAULT_FILENAME): + self._cache = default_data + self._cache_time = current_time + logger.debug(f"ProviderEnableManager: File {self.DEFAULT_FILENAME} not found, using defaults") + return default_data + + try: + # Read file via VFS + data = self.vfs.read_json(self.DEFAULT_FILENAME) + + if not isinstance(data, dict): + logger.warning(f"ProviderEnableManager: Invalid file format, using defaults") + self._cache = default_data + self._cache_time = current_time + return default_data + + # Ensure required structure + if "version" not in data: + data["version"] = self.DEFAULT_VERSION + + if "providers" not in data or not isinstance(data["providers"], dict): + data["providers"] = {} + + # Ensure all values are booleans + for provider, value in list(data["providers"].items()): + if not isinstance(value, bool): + try: + data["providers"][provider] = bool(value) + except (ValueError, TypeError): + logger.warning(f"ProviderEnableManager: Invalid value for {provider}, removing") + del data["providers"][provider] + + # Add last_updated if missing + if "last_updated" not in data: + data["last_updated"] = current_time + + # Update cache + self._cache = data + self._cache_time = current_time + + logger.debug(f"ProviderEnableManager: Loaded {len(data['providers'])} providers from file") + return data + + except Exception as e: + logger.error(f"ProviderEnableManager: Error loading {self.DEFAULT_FILENAME}: {e}") + self._cache = default_data + self._cache_time = current_time + return default_data + + def _save_file(self, data: Dict[str, Any]) -> bool: + """ + Save data to providers_enabled.json file. + + Args: + data: Data to save + + Returns: + True if successful, False otherwise + """ + try: + # Add/update metadata + data["last_updated"] = time.time() + if "version" not in data: + data["version"] = self.DEFAULT_VERSION + + # Write via VFS + success = self.vfs.write_json(self.DEFAULT_FILENAME, data) + + if success: + # Update cache + self._cache = data + self._cache_time = time.time() + logger.debug(f"ProviderEnableManager: Saved {len(data.get('providers', {}))} providers to file") + else: + logger.error(f"ProviderEnableManager: Failed to write to {self.DEFAULT_FILENAME}") + + return success + + except Exception as e: + logger.error(f"ProviderEnableManager: Error saving {self.DEFAULT_FILENAME}: {e}") + return False + + def _get_kodi_enabled_status(self, provider_name: str) -> Optional[bool]: + """ + Get enabled status from Kodi settings. + + Args: + provider_name: Provider name (e.g., "joyn_de", "rtlplus") + + Returns: + True/False if setting exists in Kodi, None if not set + """ + if not is_kodi_environment() or not self._kodi_bridge: + return None + + try: + # Parse provider_name for potential country suffix + parts = provider_name.rsplit('_', 1) + + # Check country-specific setting first (e.g., "enable_joyn_de") + if len(parts) == 2 and len(parts[1]) in (2, 3): # Country code length + provider, country = parts + country_setting = f"enable_{provider}_{country}" + + # Try country-specific setting + country_value = self._kodi_bridge.get_setting(country_setting) + if country_value: + enabled = country_value.lower() in ['true', '1', 'yes'] + logger.debug(f"ProviderEnableManager: Found Kodi setting {country_setting}={enabled}") + return enabled + + # Try general provider setting (e.g., "enable_joyn") + general_setting = f"enable_{provider_name}" + general_value = self._kodi_bridge.get_setting(general_setting) + + if general_value: + enabled = general_value.lower() in ['true', '1', 'yes'] + logger.debug(f"ProviderEnableManager: Found Kodi setting {general_setting}={enabled}") + return enabled + + # No Kodi setting found + logger.debug(f"ProviderEnableManager: No Kodi setting found for {provider_name}") + return None + + except Exception as e: + logger.warning(f"ProviderEnableManager: Error reading Kodi setting for {provider_name}: {e}") + return None + + def is_provider_enabled(self, provider_name: str) -> bool: + """ + Check if provider is enabled using precedence rules. + + Precedence: + 1. Kodi settings (if in Kodi and setting exists) + 2. providers_enabled.json file + 3. Default to True + + Args: + provider_name: Provider name (e.g., "joyn_de", "rtlplus") + + Returns: + True if enabled, False if disabled + """ + # 1. Check Kodi settings first + kodi_enabled = self._get_kodi_enabled_status(provider_name) + if kodi_enabled is not None: + return kodi_enabled + + # 2. Check file + data = self._load_file() + file_enabled = data["providers"].get(provider_name) + + if file_enabled is not None: + return file_enabled + + # 3. Default to enabled + return True + + def get_enabled_source(self, provider_name: str) -> EnableSource: + """ + Determine where the enabled status comes from. + + Args: + provider_name: Provider name + + Returns: + EnableSource enum (KODI, FILE, or DEFAULT) + """ + # Check Kodi first + kodi_enabled = self._get_kodi_enabled_status(provider_name) + if kodi_enabled is not None: + return EnableSource.KODI + + # Check file + data = self._load_file() + file_enabled = data["providers"].get(provider_name) + + if file_enabled is not None: + return EnableSource.FILE + + return EnableSource.DEFAULT + + def set_provider_enabled(self, provider_name: str, enabled: bool) -> Tuple[bool, str]: + """ + Set enabled status for a provider (writes to file only). + + Note: Cannot override Kodi settings. If provider is controlled by + Kodi settings, this will return False. + + Args: + provider_name: Provider name + enabled: True to enable, False to disable + + Returns: + Tuple of (success: bool, message: str) + """ + # Check if controlled by Kodi + source = self.get_enabled_source(provider_name) + + if source == EnableSource.KODI: + message = f"Provider '{provider_name}' is controlled by Kodi settings. " \ + f"Change the setting in Kodi addon settings." + logger.warning(f"ProviderEnableManager: {message}") + return False, message + + # Load current data + data = self._load_file() + + # Update provider status + data["providers"][provider_name] = bool(enabled) + + # Save to file + success = self._save_file(data) + + if success: + action = "enabled" if enabled else "disabled" + message = f"Provider '{provider_name}' {action} in file" + logger.info(f"ProviderEnableManager: {message}") + return True, message + else: + message = f"Failed to save enabled status for '{provider_name}'" + logger.error(f"ProviderEnableManager: {message}") + return False, message + + def get_all_enabled_status(self) -> Dict[str, Dict[str, Any]]: + """ + Get enabled status for all providers in file. + + Returns: + Dictionary mapping provider names to their status info + """ + data = self._load_file() + result = {} + + for provider_name, enabled in data["providers"].items(): + source = self.get_enabled_source(provider_name) + + result[provider_name] = { + "enabled": enabled, + "source": source.value, + "can_modify": source != EnableSource.KODI + } + + return result + + def get_provider_status(self, provider_name: str) -> Optional[Dict[str, Any]]: + """ + Get detailed status for a specific provider. + + Args: + provider_name: Provider name + + Returns: + Dictionary with status info, or None if provider not in file + """ + data = self._load_file() + + if provider_name not in data["providers"]: + return None + + enabled = data["providers"][provider_name] + source = self.get_enabled_source(provider_name) + + return { + "provider": provider_name, + "enabled": enabled, + "source": source.value, + "can_modify": source != EnableSource.KODI, + "in_file": True + } + + def delete_provider_setting(self, provider_name: str) -> Tuple[bool, str]: + """ + Delete provider setting from file (reset to default). + + Args: + provider_name: Provider name + + Returns: + Tuple of (success: bool, message: str) + """ + # Check if controlled by Kodi + source = self.get_enabled_source(provider_name) + + if source == EnableSource.KODI: + message = f"Cannot delete setting for '{provider_name}' - controlled by Kodi" + logger.warning(f"ProviderEnableManager: {message}") + return False, message + + # Load current data + data = self._load_file() + + # Remove if exists + if provider_name in data["providers"]: + del data["providers"][provider_name] + + # Save to file + success = self._save_file(data) + + if success: + message = f"Deleted setting for '{provider_name}' from file" + logger.info(f"ProviderEnableManager: {message}") + return True, message + else: + message = f"Failed to delete setting for '{provider_name}'" + logger.error(f"ProviderEnableManager: {message}") + return False, message + + # Provider not in file + message = f"Setting for '{provider_name}' not found in file" + logger.debug(f"ProviderEnableManager: {message}") + return True, message # Already not in file, so success + + def reset_all_to_defaults(self) -> Tuple[bool, str]: + """ + Reset all provider settings to defaults (empty file). + + Returns: + Tuple of (success: bool, message: str) + """ + try: + # Create default structure + default_data = { + "version": self.DEFAULT_VERSION, + "last_updated": time.time(), + "providers": {} + } + + # Save empty file + success = self._save_file(default_data) + + if success: + message = "All provider settings reset to defaults" + logger.info(f"ProviderEnableManager: {message}") + return True, message + else: + message = "Failed to reset provider settings" + logger.error(f"ProviderEnableManager: {message}") + return False, message + + except Exception as e: + message = f"Error resetting provider settings: {e}" + logger.error(f"ProviderEnableManager: {message}") + return False, message + + def get_file_info(self) -> Dict[str, Any]: + """ + Get information about the providers_enabled.json file. + + Returns: + Dictionary with file information + """ + data = self._load_file() + + return { + "filename": self.DEFAULT_FILENAME, + "version": data.get("version", self.DEFAULT_VERSION), + "last_updated": data.get("last_updated"), + "provider_count": len(data.get("providers", {})), + "file_exists": self.vfs.exists(self.DEFAULT_FILENAME), + "file_path": self.vfs.get_file_path(self.DEFAULT_FILENAME) if hasattr(self.vfs, 'get_file_path') else "vfs" + } + + def migrate_from_kodi(self) -> Tuple[bool, str, int]: + """ + Migrate all Kodi enable settings to the file. + Only copies settings that are explicitly set in Kodi. + + Returns: + Tuple of (success: bool, message: str, migrated_count: int) + """ + if not is_kodi_environment() or not self._kodi_bridge: + return False, "Not in Kodi environment or Kodi bridge not available", 0 + + try: + # Load current file data + data = self._load_file() + migrated_count = 0 + + # Discover all providers from Kodi settings + try: + kodi_providers = self._kodi_bridge.discover_all_providers() + except AttributeError: + # Fallback if discover_all_providers not available + kodi_providers = {} + + # Check each provider + for provider_name in kodi_providers: + kodi_enabled = self._get_kodi_enabled_status(provider_name) + + if kodi_enabled is not None: + # Kodi has explicit setting, migrate to file + data["providers"][provider_name] = kodi_enabled + migrated_count += 1 + logger.debug(f"ProviderEnableManager: Migrated {provider_name}={kodi_enabled}") + + if migrated_count > 0: + # Save migrated data + success = self._save_file(data) + if success: + message = f"Migrated {migrated_count} provider settings from Kodi to file" + logger.info(f"ProviderEnableManager: {message}") + return True, message, migrated_count + else: + message = "Failed to save migrated settings" + logger.error(f"ProviderEnableManager: {message}") + return False, message, 0 + else: + message = "No Kodi provider settings found to migrate" + logger.info(f"ProviderEnableManager: {message}") + return True, message, 0 + + except Exception as e: + message = f"Error migrating from Kodi: {e}" + logger.error(f"ProviderEnableManager: {message}") + return False, message, 0 \ No newline at end of file diff --git a/lib/streaming_providers/base/settings/settings_manager.py b/lib/streaming_providers/base/settings/settings_manager.py index 31a13e9..a6528f5 100644 --- a/lib/streaming_providers/base/settings/settings_manager.py +++ b/lib/streaming_providers/base/settings/settings_manager.py @@ -1319,42 +1319,19 @@ class SettingsManager: return status def is_provider_enabled(self, provider_name: str, country: Optional[str] = None) -> bool: - """ - Check if a provider is enabled via Kodi settings. + """Check if provider is enabled (using ProviderEnableManager)""" + # Import here to avoid circular imports + from .provider_enable_manager import ProviderEnableManager - Args: - provider_name: Name of the provider (e.g., 'joyn', 'magenta2') - country: Optional country code (e.g., 'de', 'at') + enable_manager = ProviderEnableManager(self.config_dir_path) - Returns: - True if provider is enabled, False if disabled, True by default if no setting found - """ - if not self.kodi_bridge or not self.kodi_bridge.is_kodi_environment(): - return True # Default to enabled if not in Kodi environment + # Handle country parameter if provider_name doesn't have suffix + if country and '_' not in provider_name: + provider_to_check = f"{provider_name}_{country}" + else: + provider_to_check = provider_name - try: - # Priority 1: Check provider_country specific enable setting (e.g., enable_joyn_de) - if country: - enable_setting_country = f"enable_{provider_name}_{country}" - if self.kodi_bridge.addon.getSetting(enable_setting_country): - enabled = self.kodi_bridge.addon.getSetting(enable_setting_country) - logger.debug(f"Found country-specific enable setting '{enable_setting_country}': {enabled}") - return enabled.lower() in ['true', '1', 'yes'] - - # Priority 2: Check general provider enable setting (e.g., enable_joyn) - enable_setting_general = f"enable_{provider_name}" - if self.kodi_bridge.addon.getSetting(enable_setting_general): - enabled = self.kodi_bridge.addon.getSetting(enable_setting_general) - logger.debug(f"Found general enable setting '{enable_setting_general}': {enabled}") - return enabled.lower() in ['true', '1', 'yes'] - - # No enable setting found - default to enabled - logger.debug(f"No enable setting found for '{provider_name}', defaulting to enabled") - return True - - except Exception as e: - logger.warning(f"Error checking enable status for '{provider_name}': {e}, defaulting to enabled") - return True + return enable_manager.is_provider_enabled(provider_to_check) # ============= API Import Methods ============= diff --git a/service.py b/service.py index bb2326c..59f020f 100644 --- a/service.py +++ b/service.py @@ -20,6 +20,7 @@ try: from streaming_providers.base.utils import logger, MPDRewriter, MPDCacheManager from streaming_providers.base.utils.environment import get_environment_manager, get_vfs_instance from streaming_providers.base.utils.environment import is_kodi_environment + from streaming_providers.base.settings.provider_enable_manager import ProviderEnableManager except ImportError as import_err: print(f"Ultimate Backend: Critical import failed - {str(import_err)}", file=sys.stderr) raise @@ -425,10 +426,16 @@ class UltimateService: except json.JSONDecodeError: # If not JSON, try to parse as query string try: - headers_dict = dict(parse_qsl(req_headers)) + # parse_qsl can raise ValueError if the query string is malformed + parsed_items = parse_qsl(req_headers) + headers_dict = dict(parsed_items) return urlencode(headers_dict) - except: - logger.warning(f"Failed to parse headers: {req_headers}") + except ValueError as val_err: + logger.warning(f"Invalid query string format in headers: {val_err}") + return req_headers + except Exception as parse_err: + # Catch anything else unexpected + logger.warning(f"Unexpected error parsing headers: {parse_err}") return req_headers else: # Assume it's already URL-encoded or query string format @@ -1608,6 +1615,120 @@ class UltimateService: response.status = 500 return {'error': f'Internal server error: {str(e)}'} + # Add to service.py setup_routes() + + @self.app.route('/api/providers/enabled', method='GET') + def get_all_enabled_status(): + """Get enabled status for all providers""" + try: + manager = self.manager # ProviderManager + enable_manager = ProviderEnableManager() # Our new class + + all_providers = manager.list_providers() + result = {} + + for provider in all_providers: + # Get current status (following precedence) + status = enable_manager.is_provider_enabled(provider) + + # Get source information + source = enable_manager.get_enabled_source(provider) + + result[provider] = { + 'enabled': status, + 'source': source, # 'kodi', 'file', or 'default' + 'can_modify': source != 'kodi' # Can't modify if set in Kodi + } + + return { + 'success': True, + 'providers': result, + 'count': len(result) + } + + except Exception as e: + logger.error(f"Error getting enabled status: {e}") + response.status = 500 + return {'error': str(e)} + + @self.app.route('/api/providers//enabled', method='GET') + def get_provider_enabled(provider): + """Get enabled status for specific provider""" + try: + # Validate provider exists + if not self.manager.get_provider(provider): + response.status = 404 + return {'error': f'Provider {provider} not found'} + + enable_manager = ProviderEnableManager() + status = enable_manager.is_provider_enabled(provider) + source = enable_manager.get_enabled_source(provider) + + return { + 'success': True, + 'provider': provider, + 'enabled': status, + 'source': source, + 'can_modify': source != 'kodi' + } + + except Exception as e: + logger.error(f"Error getting enabled status for {provider}: {e}") + response.status = 500 + return {'error': str(e)} + + @self.app.route('/api/providers//enabled', method='POST') + def set_provider_enabled(provider): + """Set enabled status for provider (writes to file only)""" + try: + # Validate provider exists + if not self.manager.get_provider(provider): + response.status = 404 + return {'error': f'Provider {provider} not found'} + + # Parse request + try: + data = request.json + if not data or 'enabled' not in data: + response.status = 400 + return {'error': 'Missing "enabled" field'} + + enabled = bool(data['enabled']) + except ValueError: + response.status = 400 + return {'error': 'Invalid JSON'} + + # Check if controlled by Kodi + enable_manager = ProviderEnableManager() + source = enable_manager.get_enabled_source(provider) + + if source == 'kodi': + response.status = 403 + return { + 'error': f'Provider {provider} is controlled by Kodi settings', + 'hint': 'Change the setting in Kodi addon settings' + } + + # Write to file + success = enable_manager.set_provider_enabled(provider, enabled) + + if success: + return { + 'success': True, + 'provider': provider, + 'enabled': enabled, + 'source': 'file', + 'message': f'Provider {provider} {"enabled" if enabled else "disabled"} in file' + } + else: + response.status = 500 + return {'error': 'Failed to save setting'} + + except Exception as e: + logger.error(f"Error setting enabled status for {provider}: {e}") + response.status = 500 + return {'error': str(e)} + @self.app.route('/api/config/export') def export_config(): """Export all configurations as JSON""" @@ -1635,65 +1756,93 @@ class UltimateService: """Import configurations from JSON""" try: import_data = request.json - if not import_data: - response.status = 400 - return {'error': 'No data provided'} + except ValueError: + response.status = 400 + return {'error': 'Invalid JSON format'} - # First save the import data to a temp file + if not import_data: + response.status = 400 + return {'error': 'No data provided'} + + # Validate it's a dict + if not isinstance(import_data, dict): + response.status = 400 + return {'error': 'Import data must be a JSON object'} + + # Create temp file + try: import tempfile import uuid + import json temp_dir = tempfile.gettempdir() temp_file = os.path.join(temp_dir, f'import_{uuid.uuid4()}.json') with open(temp_file, 'w', encoding='utf-8') as f: json.dump(import_data, f) + except (IOError, OSError, PermissionError) as file_err: + logger.error(f"Failed to create temp file: {file_err}") + response.status = 500 + return {'error': 'Failed to process import file'} + imported_count = 0 + try: # Use SettingsManager to import settings_manager = self._get_settings_manager() - # Since SettingsManager doesn't have an import method, we'll process it manually # Import credentials credentials = import_data.get('providers', {}) - imported_count = 0 for provider_name, provider_data in credentials.items(): + # Validate provider data + if not isinstance(provider_data, dict): + logger.warning(f"Skipping invalid provider data for {provider_name}") + continue + # Extract credential data if available if 'credentials' in provider_data: cred_data = provider_data['credentials'] - success, message = settings_manager.save_provider_credentials_from_api( - provider_name, cred_data - ) - if success: - imported_count += 1 - logger.info(f"Imported credentials for {provider_name}") + if isinstance(cred_data, dict): + success, message = settings_manager.save_provider_credentials_from_api( + provider_name, cred_data + ) + if success: + imported_count += 1 + logger.info(f"Imported credentials for {provider_name}") + else: + logger.warning(f"Failed to import credentials for {provider_name}: {message}") # Import proxy data if available if 'proxy' in provider_data: proxy_data = provider_data['proxy'] - success, message = settings_manager.save_provider_proxy_from_api( - provider_name, proxy_data - ) - if success: - imported_count += 1 - logger.info(f"Imported proxy for {provider_name}") + if isinstance(proxy_data, dict): + success, message = settings_manager.save_provider_proxy_from_api( + provider_name, proxy_data + ) + if success: + imported_count += 1 + logger.info(f"Imported proxy for {provider_name}") + else: + logger.warning(f"Failed to import proxy for {provider_name}: {message}") - # Clean up temp file - try: - os.remove(temp_file) - except: - pass - - return { - 'success': True, - 'imported': imported_count, - 'message': f'Imported {imported_count} configurations' - } - - except Exception as e: - logger.error(f"Error importing config: {e}") + except Exception as process_err: + logger.error(f"Error during import processing: {process_err}", exc_info=True) response.status = 500 - return {'error': str(e)} + return {'error': f'Import failed: {str(process_err)}'} + + finally: + # Always try to clean up temp file + try: + if os.path.exists(temp_file): + os.remove(temp_file) + except (FileNotFoundError, PermissionError, OSError) as cleanup_err: + logger.debug(f"Could not remove temp file {temp_file}: {cleanup_err}") + + return { + 'success': True, + 'imported': imported_count, + 'message': f'Imported {imported_count} configurations' + } @self.app.route('/config') def serve_config_ui():