mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-09-16 06:02:35 +02:00
Add web config
This commit is contained in:
Nirvana
@@ -14,7 +14,6 @@ services:
|
||||
- "7777:7777"
|
||||
environment:
|
||||
- ULTIMATE_PORT=7777
|
||||
- ULTIMATE_COUNTRY=DE
|
||||
- ULTIMATE_DEBUG=false
|
||||
- TZ=${TZ:-Europe/Berlin}
|
||||
volumes:
|
||||
|
||||
@@ -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
|
||||
@@ -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 =============
|
||||
|
||||
|
||||
+185
-36
@@ -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/<provider>/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/<provider>/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():
|
||||
|
||||
Reference in New Issue
Block a user