nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-24 18:12:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Dockerize addon

Browse Source
This commit is contained in:
Nirvana
2025-12-10 13:56:11 +01:00
parent 97e1593400
commit e4c5a69a60
11 changed files with 1633 additions and 237 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
+41
View File
@@ -0,0 +1,41 @@
# Git
.git
.gitignore
.gitattributes
# IDE
.vscode
.idea
*.swp
*.swo
# OS
.DS_Store
Thumbs.db
# Python
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
.env
.venv
venv/
ENV/
env/
# Logs
logs/
*.log
# Test and coverage
.coverage
htmlcov/
.pytest_cache/
.tox/
# Distribution
dist/
build/
*.egg-info/
Dockerfile
+80
View File
@@ -0,0 +1,80 @@
# Use Python 3.11 slim as base
FROM python:3.11-slim
# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
TZ=UTC \
ULTIMATE_PORT=7777 \
ULTIMATE_COUNTRY=DE \
PYTHONPATH=/app/lib:$PYTHONPATH
# Create non-root user
RUN useradd -m -u 1000 -s /bin/bash kodi
# Install system dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
libxml2-dev \
libxslt-dev \
libffi-dev \
libssl-dev \
libxmlsec1-dev \
pkg-config \
curl \
&& rm -rf /var/lib/apt/lists/*
# Set working directory
WORKDIR /app
# Copy entire project
COPY . /app/
# Create config directory and ensure proper permissions
RUN mkdir -p /config && \
chown -R kodi:kodi /app /config
USER kodi
# Install Python dependencies
RUN pip install --no-cache-dir --user \
bottle \
requests \
lxml \
defusedxml \
m3u8 \
iso8601 \
pycountry \
python-dateutil \
urllib3 \
chardet \
pycryptodome \
cryptography
# Create requirements.txt for documentation
RUN echo "bottle>=0.12.25" > /app/requirements.txt && \
echo "requests>=2.31.0" >> /app/requirements.txt && \
echo "lxml>=4.9.3" >> /app/requirements.txt && \
echo "defusedxml>=0.7.1" >> /app/requirements.txt && \
echo "m3u8>=4.0.0" >> /app/requirements.txt && \
echo "iso8601>=1.1.0" >> /app/requirements.txt && \
echo "pycountry>=23.12.11" >> /app/requirements.txt && \
echo "python-dateutil>=2.8.2" >> /app/requirements.txt && \
echo "urllib3>=2.0.7" >> /app/requirements.txt && \
echo "chardet>=5.2.0" >> /app/requirements.txt && \
echo "pycryptodome>=3.19.0" >> /app/requirements.txt && \
echo "cryptography>=41.0.0" >> /app/requirements.txt
# Expose the service port
EXPOSE ${ULTIMATE_PORT}
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
CMD curl -f http://localhost:${ULTIMATE_PORT}/api/providers || exit 1
# Run in standalone mode (since Kodi isn't available in Docker)
ENTRYPOINT ["python", "/app/service.py"]
# Default command
CMD ["--standalone", "--config-dir", "/config"]
docker-compose.yml
+28
View File
@@ -0,0 +1,28 @@
version: '3.8'
services:
ultimate-backend:
build: .
container_name: ultimate-backend
restart: unless-stopped
ports:
- "7777:7777"
environment:
- ULTIMATE_PORT=7777
- ULTIMATE_COUNTRY=DE
- TZ=Europe/Berlin
- DEBUG_MODE=true
volumes:
- ./config:/config
- ./logs:/logs
- ./cache:/cache
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:7777/api/providers"]
interval: 30s
timeout: 3s
retries: 3
start_period: 5s
labels:
- "traefik.enable=true"
- "traefik.http.routers.ultimate.rule=Host(`ultimate.local`)"
- "traefik.http.services.ultimate.loadbalancer.server.port=7777"
lib/streaming_providers/base/settings/kodi_settings_bridge.py
+138 -78
View File
@@ -1,17 +1,12 @@
# streaming_providers/base/settings/kodi_settings_bridge.py
from typing import Dict, List, Optional, Set, Tuple
from typing import Dict, List, Optional, Set, Tuple, Any
import json
import xml.etree.ElementTree as ElementTree
from ..auth.credentials import BaseCredentials, UserPasswordCredentials, ClientCredentials
from ..models.proxy_models import ProxyConfig
from ..utils.logger import logger
try:
import xbmcaddon
KODI_AVAILABLE = True
except ImportError:
KODI_AVAILABLE = False
from ..utils.environment import get_environment_manager, is_kodi_environment, get_vfs_instance
class KodiSettingsBridge:
@@ -24,13 +19,19 @@ class KodiSettingsBridge:
"""Initialize Kodi settings bridge"""
self.addon = None
self.addon_id = addon_id
self._env_manager = get_environment_manager()
# Initialize VFS with config directory support
from ..utils.vfs import VFS
self.vfs = VFS(config_dir=config_dir)
self.vfs = get_vfs_instance(config_dir=config_dir)
if KODI_AVAILABLE:
# Settings storage for standalone mode
self._standalone_settings: Dict[str, str] = {}
self._settings_file = "standalone_settings.json"
self._load_standalone_settings()
if is_kodi_environment():
try:
import xbmcaddon
if addon_id:
self.addon = xbmcaddon.Addon(addon_id)
else:
@@ -43,14 +44,72 @@ class KodiSettingsBridge:
def is_kodi_environment(self) -> bool:
"""Check if currently running in Kodi environment"""
return KODI_AVAILABLE and self.addon is not None
return is_kodi_environment() and self.addon is not None
def _load_standalone_settings(self) -> None:
"""Load settings from file in standalone mode"""
if not is_kodi_environment() and self.vfs.exists(self._settings_file):
try:
content = self.vfs.read_text(self._settings_file)
if content:
self._standalone_settings = json.loads(content)
logger.debug(f"Loaded {len(self._standalone_settings)} standalone settings")
except Exception as e:
logger.error(f"Error loading standalone settings: {e}")
def _save_standalone_settings(self) -> None:
"""Save settings to file in standalone mode"""
if not is_kodi_environment():
try:
self.vfs.write_json(self._settings_file, self._standalone_settings)
logger.debug(f"Saved {len(self._standalone_settings)} standalone settings")
except Exception as e:
logger.error(f"Error saving standalone settings: {e}")
def get_setting(self, setting_id: str, default: str = "") -> str:
"""Get setting value, works in both Kodi and standalone mode"""
if self.is_kodi_environment():
try:
return self.addon.getSetting(setting_id) or default
except Exception as e:
logger.error(f"Error getting setting {setting_id}: {e}")
return default
else:
# Standalone mode: use local storage
return self._standalone_settings.get(setting_id, default)
def set_setting(self, setting_id: str, value: str) -> bool:
"""Set setting value, works in both Kodi and standalone mode"""
if self.is_kodi_environment():
try:
self.addon.setSetting(setting_id, value)
logger.debug(f"Set Kodi setting {setting_id}")
return True
except Exception as e:
logger.error(f"Error setting {setting_id}: {e}")
return False
else:
# Standalone mode: update local storage
self._standalone_settings[setting_id] = value
self._save_standalone_settings()
logger.debug(f"Set standalone setting {setting_id}")
return True
def get_addon_info(self) -> Dict[str, str]:
"""Get information about current Kodi addon (ID, version, etc.)"""
"""Get information about current Kodi addon or environment"""
if not self.is_kodi_environment():
return {"error": "Not in Kodi environment"}
# Return environment info in standalone mode
return {
'environment': 'standalone',
'id': self._env_manager.get_config('addon_id', 'standalone'),
'name': self._env_manager.get_config('addon_name', 'Ultimate Backend'),
'version': self._env_manager.get_config('addon_version', '1.0.0'),
'config_dir': self.vfs.base_path,
'profile_path': self._env_manager.get_config('profile_path', ''),
}
return {
'environment': 'kodi',
'id': self.addon.getAddonInfo('id'),
'name': self.addon.getAddonInfo('name'),
'version': self.addon.getAddonInfo('version'),
@@ -68,7 +127,8 @@ class KodiSettingsBridge:
List of all setting IDs found in settings.xml
"""
if not self.is_kodi_environment():
return []
# In standalone mode, return keys from standalone settings
return list(self._standalone_settings.keys())
try:
# Read settings.xml from VFS base path (addon profile directory)
@@ -76,9 +136,6 @@ class KodiSettingsBridge:
if not xml_content:
logger.warning("settings.xml not found or empty")
return []
if not xml_content:
logger.warning("settings.xml is empty")
return []
root = ElementTree.fromstring(xml_content)
@@ -141,16 +198,15 @@ class KodiSettingsBridge:
def discover_all_providers(self) -> Dict[str, List[str]]:
"""
Scan all Kodi settings and discover providers with their countries dynamically.
Scan all settings and discover providers with their countries dynamically.
Works in both Kodi and standalone mode.
Returns:
Dict mapping provider names to list of countries.
Empty list means provider without country.
Example: {'joyn': ['de', 'at'], 'rtlplus': ['de'], 'zattoo': []}
"""
if not self.is_kodi_environment():
return {}
discovered: Dict[str, Set[Optional[str]]] = {}
# Get all setting IDs
@@ -218,7 +274,7 @@ class KodiSettingsBridge:
available_countries: List of country codes to check
Returns:
List of country codes that have credentials in Kodi
List of country codes that have credentials
"""
discovered_countries = self.detect_countries_for_provider(provider)
return [c for c in available_countries if c in discovered_countries]
@@ -227,23 +283,21 @@ class KodiSettingsBridge:
def read_credentials_from_kodi(self, provider: str, country: Optional[str] = None) -> Optional[BaseCredentials]:
"""
Read authentication credentials from Kodi settings for a provider.
Read authentication credentials from settings for a provider.
Uses convention: {provider}_{country}_username, {provider}_{country}_password, etc.
If country is None, tries without country suffix for backward compatibility.
Works in both Kodi and standalone mode.
"""
if not self.is_kodi_environment():
return None
country_suffix = f"_{country}" if country else ""
try:
# Try convention-based setting names
username = self.addon.getSetting(f'{provider}{country_suffix}_username')
password = self.addon.getSetting(f'{provider}{country_suffix}_password')
client_id = self.addon.getSetting(f'{provider}{country_suffix}_client_id')
client_secret = self.addon.getSetting(f'{provider}{country_suffix}_client_secret')
# Use unified get_setting method
username = self.get_setting(f'{provider}{country_suffix}_username')
password = self.get_setting(f'{provider}{country_suffix}_password')
client_id = self.get_setting(f'{provider}{country_suffix}_client_id')
client_secret = self.get_setting(f'{provider}{country_suffix}_client_secret')
logger.debug(f"Kodi settings for {provider}{country_suffix}:")
logger.debug(f"Settings for {provider}{country_suffix}:")
logger.debug(f" username: '{username}' (empty={not username})")
logger.debug(f" password: {'***' if password else '(empty)'}")
logger.debug(f" client_id: '{client_id}' (empty={not client_id})")
@@ -251,58 +305,55 @@ class KodiSettingsBridge:
# Determine credential type based on available values
if username and password:
logger.info(f"Found username/password credentials for {provider}{country_suffix} in Kodi")
logger.info(f"Found username/password credentials for {provider}{country_suffix}")
return UserPasswordCredentials(
username=username.strip(),
password=password.strip(),
client_id=client_id.strip() if client_id else None
)
elif client_id and client_secret:
logger.info(f"Found client credentials for {provider}{country_suffix} in Kodi")
logger.info(f"Found client credentials for {provider}{country_suffix}")
return ClientCredentials(
client_id=client_id.strip(),
client_secret=client_secret.strip()
)
logger.debug(f"No valid credentials found in Kodi for {provider}{country_suffix}")
logger.debug(f"No valid credentials found for {provider}{country_suffix}")
return None
except Exception as e:
logger.error(f"Error reading credentials from Kodi for {provider}: {e}")
logger.error(f"Error reading credentials for {provider}: {e}")
return None
def write_credentials_to_kodi(self, provider: str, credentials: BaseCredentials,
country: Optional[str] = None) -> bool:
"""Write authentication credentials to Kodi settings"""
if not self.is_kodi_environment():
return False
"""Write authentication credentials to settings"""
country_suffix = f"_{country}" if country else ""
try:
if isinstance(credentials, UserPasswordCredentials):
self.addon.setSetting(f'{provider}{country_suffix}_username', credentials.username)
self.addon.setSetting(f'{provider}{country_suffix}_password', credentials.password)
self.set_setting(f'{provider}{country_suffix}_username', credentials.username)
self.set_setting(f'{provider}{country_suffix}_password', credentials.password)
if credentials.client_id:
self.addon.setSetting(f'{provider}{country_suffix}_client_id', credentials.client_id)
logger.info(f"Wrote username/password credentials to Kodi for {provider}{country_suffix}")
self.set_setting(f'{provider}{country_suffix}_client_id', credentials.client_id)
logger.info(f"Wrote username/password credentials for {provider}{country_suffix}")
return True
elif isinstance(credentials, ClientCredentials):
self.addon.setSetting(f'{provider}{country_suffix}_client_id', credentials.client_id)
self.addon.setSetting(f'{provider}{country_suffix}_client_secret', credentials.client_secret)
logger.info(f"Wrote client credentials to Kodi for {provider}{country_suffix}")
self.set_setting(f'{provider}{country_suffix}_client_id', credentials.client_id)
self.set_setting(f'{provider}{country_suffix}_client_secret', credentials.client_secret)
logger.info(f"Wrote client credentials for {provider}{country_suffix}")
return True
return False
except Exception as e:
logger.error(f"Error writing credentials to Kodi for {provider}: {e}")
logger.error(f"Error writing credentials for {provider}: {e}")
return False
def sync_credentials_to_file(self, provider: str, credential_manager,
country: Optional[str] = None) -> bool:
"""Sync provider credentials from Kodi settings to credential file"""
"""Sync provider credentials from settings to credential file"""
credentials = self.read_credentials_from_kodi(provider, country)
if not credentials:
logger.debug(f"No credentials to sync for {provider}")
@@ -316,32 +367,29 @@ class KodiSettingsBridge:
success = credential_manager.save_credentials(provider, credentials, country)
if success:
logger.info(f"Synced credentials from Kodi to file for {provider}")
logger.info(f"Synced credentials from settings to file for {provider}")
return success
# ============= Proxy Operations =============
def read_proxy_config_from_kodi(self, provider: str, country: Optional[str] = None) -> Optional[ProxyConfig]:
"""
Read proxy configuration from Kodi settings for a provider.
Read proxy configuration from settings for a provider.
Uses convention: {provider}_{country}_proxy_enabled, {provider}_{country}_proxy_host, etc.
"""
if not self.is_kodi_environment():
return None
country_suffix = f"_{country}" if country else ""
try:
# Check if proxy is enabled
proxy_enabled = self.addon.getSetting(f'{provider}{country_suffix}_proxy_enabled')
proxy_enabled = self.get_setting(f'{provider}{country_suffix}_proxy_enabled')
logger.debug(f"Proxy enabled setting for {provider}{country_suffix}: '{proxy_enabled}'")
if not proxy_enabled or proxy_enabled.lower() not in ['true', '1', 'yes']:
logger.debug(f"Proxy not enabled for {provider}{country_suffix}")
return None
proxy_host = self.addon.getSetting(f'{provider}{country_suffix}_proxy_host')
proxy_port_str = self.addon.getSetting(f'{provider}{country_suffix}_proxy_port')
proxy_host = self.get_setting(f'{provider}{country_suffix}_proxy_host')
proxy_port_str = self.get_setting(f'{provider}{country_suffix}_proxy_port')
logger.debug(f"Proxy settings for {provider}{country_suffix}:")
logger.debug(f" host: '{proxy_host}'")
@@ -360,36 +408,33 @@ class KodiSettingsBridge:
# Create proxy config
proxy_config = ProxyConfig(host=proxy_host.strip(), port=proxy_port)
logger.info(f"Found proxy config for {provider}{country_suffix} in Kodi: {proxy_host}:{proxy_port}")
logger.info(f"Found proxy config for {provider}{country_suffix}: {proxy_host}:{proxy_port}")
return proxy_config
except Exception as e:
logger.error(f"Error reading proxy config from Kodi for {provider}: {e}")
logger.error(f"Error reading proxy config for {provider}: {e}")
return None
def write_proxy_config_to_kodi(self, provider: str, proxy_config: ProxyConfig,
country: Optional[str] = None) -> bool:
"""Write proxy configuration to Kodi settings"""
if not self.is_kodi_environment():
return False
"""Write proxy configuration to settings"""
country_suffix = f"_{country}" if country else ""
try:
self.addon.setSetting(f'{provider}{country_suffix}_proxy_enabled', 'true')
self.addon.setSetting(f'{provider}{country_suffix}_proxy_host', proxy_config.host)
self.addon.setSetting(f'{provider}{country_suffix}_proxy_port', str(proxy_config.port))
self.set_setting(f'{provider}{country_suffix}_proxy_enabled', 'true')
self.set_setting(f'{provider}{country_suffix}_proxy_host', proxy_config.host)
self.set_setting(f'{provider}{country_suffix}_proxy_port', str(proxy_config.port))
logger.info(f"Wrote proxy config to Kodi for {provider}{country_suffix}")
logger.info(f"Wrote proxy config for {provider}{country_suffix}")
return True
except Exception as e:
logger.error(f"Error writing proxy config to Kodi for {provider}: {e}")
logger.error(f"Error writing proxy config for {provider}: {e}")
return False
def sync_proxy_config_to_file(self, provider: str, proxy_manager,
country: Optional[str] = None) -> bool:
"""Sync provider proxy config from Kodi settings to proxy config file"""
"""Sync provider proxy config from settings to proxy config file"""
proxy_config = self.read_proxy_config_from_kodi(provider, country)
if not proxy_config:
logger.debug(f"No proxy config to sync for {provider}")
@@ -403,12 +448,12 @@ class KodiSettingsBridge:
success = proxy_manager.set_proxy_config(provider, proxy_config, country)
if success:
logger.info(f"Synced proxy config from Kodi to file for {provider}")
logger.info(f"Synced proxy config from settings to file for {provider}")
return success
def read_ip_address_from_kodi(self, provider: str, country: Optional[str] = None) -> Optional[str]:
"""
Read IP address from Kodi settings for a provider.
Read IP address from settings for a provider.
Uses convention: {provider}_{country}_ipaddress or {provider}_ipaddress
Args:
@@ -418,13 +463,10 @@ class KodiSettingsBridge:
Returns:
Configured IP address or None if not set
"""
if not self.is_kodi_environment():
return None
country_suffix = f"_{country}" if country else ""
try:
ip_address = self.addon.getSetting(f'{provider}{country_suffix}_ipaddress')
ip_address = self.get_setting(f'{provider}{country_suffix}_ipaddress')
logger.debug(f"IP address setting for {provider}{country_suffix}: '{ip_address}'")
@@ -436,7 +478,7 @@ class KodiSettingsBridge:
return None
except Exception as e:
logger.error(f"Error reading IP address from Kodi for {provider}: {e}")
logger.error(f"Error reading IP address for {provider}: {e}")
return None
# ============= Comparison Helpers =============
@@ -490,4 +532,22 @@ class KodiSettingsBridge:
proxy1.scope.license != proxy2.scope.license):
return False
return True
return True
def debug_info(self) -> Dict[str, Any]:
"""Get debug information about the settings bridge"""
info = {
'environment': 'kodi' if self.is_kodi_environment() else 'standalone',
'addon_info': self.get_addon_info(),
'has_addon': self.addon is not None,
'standalone_settings_count': len(self._standalone_settings),
'vfs_base_path': self.vfs.base_path,
}
if self.is_kodi_environment():
info['addon_id'] = self.addon_id
info['kodi_available'] = True
else:
info['kodi_available'] = False
return info
lib/streaming_providers/base/settings/settings_manager.py
+418 -2
View File
@@ -7,13 +7,13 @@ Supports country-specific credentials, sessions, and configurations
import json
import time
import os
from typing import Dict, Any, Optional, List
from typing import Dict, Any, Optional, List, Tuple
from pathlib import Path
from dataclasses import dataclass, field
from ..auth.session_manager import SessionManager
from ..auth.credential_manager import CredentialManager
from ..auth.credentials import BaseCredentials
from ..auth.credentials import BaseCredentials, ClientCredentials, UserPasswordCredentials
from ..network.proxy_manager import ProxyConfigManager
from ..models.proxy_models import ProxyConfig
from ..utils.logger import logger
@@ -1356,6 +1356,422 @@ class SettingsManager:
logger.warning(f"Error checking enable status for '{provider_name}': {e}, defaulting to enabled")
return True
# ============= API Import Methods =============
@staticmethod
def parse_provider_country(provider_name: str) -> Tuple[str, Optional[str]]:
"""
Parse provider_country format into (provider, country)
Examples:
"joyn_de" → ("joyn", "de")
"magenta2_de" → ("magenta2", "de")
"some_provider" → ("some_provider", None)
Args:
provider_name: Combined provider name like "joyn_de" or plain "provider"
Returns:
Tuple of (provider, country) where country may be None
"""
# Split by last underscore
parts = provider_name.rsplit('_', 1)
# Check if last part looks like a country code (2-3 alphabetic chars)
if len(parts) == 2 and len(parts[1]) in (2, 3) and parts[1].isalpha():
return parts[0], parts[1].lower()
# No country suffix detected
return provider_name, None
def get_auth_status(self, provider_name: str) -> Dict[str, Any]:
"""
Get comprehensive authentication status for a provider
Determines the "real" authentication state accounting for:
- Providers with credentials but no token (lazy auth)
- Expired vs valid tokens
- Different credential types
Args:
provider_name: Provider name, optionally with country (e.g., "joyn_de")
Returns:
Dictionary with authentication status information
"""
# Parse provider and country
provider, country = self.parse_provider_country(provider_name)
# Check if provider is registered
if not self.is_provider_registered(provider):
return {
'provider': provider_name,
'auth_state': 'not_authenticated',
'error': f'Provider "{provider}" is not registered',
'is_ready': False
}
# Load credentials
credentials = self.get_provider_credentials(provider, country)
has_credentials = credentials is not None and credentials.validate()
# Load token data
token_data = self.load_token_data(provider, country)
has_token = token_data is not None
# Check token expiration (using same buffer as SessionManager)
token_valid = False
token_expires_at = None
if has_token:
# Use SessionManager's expiration check
token_valid = not self.session_manager._is_token_expired(token_data, buffer_seconds=300)
# Calculate expiration timestamp
if 'issued_at' in token_data and 'expires_in' in token_data:
token_expires_at = token_data['issued_at'] + token_data['expires_in']
# Determine authentication state
auth_state = self._determine_auth_state(
has_credentials=has_credentials,
has_token=has_token,
token_valid=token_valid,
credential_type=credentials.credential_type if credentials else None
)
# Build response
status = {
'provider': provider_name,
'auth_state': auth_state,
'credential_type': credentials.credential_type if credentials else None,
'has_credentials': has_credentials,
'has_active_token': token_valid,
'token_expires_at': token_expires_at,
'is_ready': auth_state in ['user_authenticated', 'client_authenticated']
}
return status
@staticmethod
def _determine_auth_state(has_credentials: bool, has_token: bool,
token_valid: bool, credential_type: Optional[str]) -> str:
"""
Determine authentication state from credential and token status
Args:
has_credentials: Whether valid credentials exist
has_token: Whether token data exists
token_valid: Whether token is valid (not expired)
credential_type: Type of credential ("user_password" or "client_credentials")
Returns:
One of: "not_authenticated", "credentials_only", "user_authenticated", "client_authenticated"
"""
# No credentials and no token
if not has_credentials and not has_token:
return "not_authenticated"
# No credentials but has token (shouldn't happen, but handle it)
if not has_credentials and has_token:
return "not_authenticated"
# Has credentials but no token (lazy auth - hasn't authenticated yet)
if has_credentials and not has_token:
return "credentials_only"
# Has credentials and token but token expired (needs re-auth)
if has_credentials and has_token and not token_valid:
return "credentials_only"
# Has credentials and valid token - determine type
if has_credentials and has_token and token_valid:
if credential_type == "user_password":
return "user_authenticated"
elif credential_type == "client_credentials":
return "client_authenticated"
else:
# Unknown credential type but authenticated
return "user_authenticated"
# Fallback
return "not_authenticated"
def save_provider_credentials_from_api(self, provider_name: str,
credentials_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
Save credentials from API request (bypasses Kodi sync)
Args:
provider_name: Provider name, optionally with country (e.g., "joyn_de")
credentials_data: Dictionary with credential data
For user_password: {"username": "...", "password": "...", "client_id": "..." (optional)}
For client_credentials: {"client_id": "...", "client_secret": "..."}
Returns:
Tuple of (success: bool, message: str)
"""
try:
# Parse provider and country
provider, country = self.parse_provider_country(provider_name)
# Check if provider is registered
if not self.is_provider_registered(provider):
return False, f'Provider "{provider}" is not registered'
# Validate credentials_data format
if not isinstance(credentials_data, dict):
return False, "Credentials data must be a dictionary"
# Determine credential type and create appropriate object
credentials = self._parse_credentials_from_dict(credentials_data)
if not credentials:
return False, "Invalid credentials format. Must provide either (username + password) or (client_id + client_secret)"
# Validate credentials
if not credentials.validate():
return False, f"Credential validation failed: missing required fields"
# Save directly to file (bypass Kodi sync)
success = self.credential_manager.save_credentials(provider, credentials, country)
if success:
country_str = f" ({country})" if country else ""
logger.info(f"Successfully saved credentials from API for {provider}{country_str}")
return True, "Credentials saved successfully"
else:
return False, "Failed to save credentials to file"
except Exception as e:
logger.error(f"Error saving credentials from API for {provider_name}: {e}")
return False, f"Internal error: {str(e)}"
@staticmethod
def _parse_credentials_from_dict(credentials_data: Dict[str, Any]) -> Optional[BaseCredentials]:
"""
Parse credentials dictionary and create appropriate credential object
Args:
credentials_data: Dictionary with credential data
Returns:
BaseCredentials instance or None if invalid
"""
# Check for user_password credentials
has_username = 'username' in credentials_data and credentials_data['username']
has_password = 'password' in credentials_data and credentials_data['password']
if has_username and has_password:
return UserPasswordCredentials(
username=credentials_data['username'].strip(),
password=credentials_data['password'].strip(),
client_id=credentials_data.get('client_id', '').strip() or None
)
# Check for client credentials
has_client_id = 'client_id' in credentials_data and credentials_data['client_id']
has_client_secret = 'client_secret' in credentials_data and credentials_data['client_secret']
if has_client_id and has_client_secret:
return ClientCredentials(
client_id=credentials_data['client_id'].strip(),
client_secret=credentials_data['client_secret'].strip()
)
# Invalid format
return None
def save_provider_proxy_from_api(self, provider_name: str,
proxy_data: Dict[str, Any]) -> Tuple[bool, str]:
"""
Save proxy configuration from API request (bypasses Kodi sync)
Args:
provider_name: Provider name, optionally with country (e.g., "joyn_de")
proxy_data: Dictionary with proxy data
Required: {"host": "...", "port": 8080}
Optional: {"proxy_type": "http", "username": "...", "password": "...",
"timeout": 30, "verify_ssl": true}
Returns:
Tuple of (success: bool, message: str)
"""
try:
# Parse provider and country
provider, country = self.parse_provider_country(provider_name)
# Check if provider is registered
if not self.is_provider_registered(provider):
return False, f'Provider "{provider}" is not registered'
# Validate proxy_data format
if not isinstance(proxy_data, dict):
return False, "Proxy data must be a dictionary"
# Create ProxyConfig object
proxy_config = self._parse_proxy_from_dict(proxy_data)
if not proxy_config:
return False, "Invalid proxy format. Must provide 'host' and 'port'"
# Validate proxy config
if not proxy_config.validate():
return False, "Proxy validation failed: invalid host, port, or timeout"
# Save directly to file (bypass Kodi sync)
success = self.proxy_manager.set_proxy_config(provider, proxy_config, country)
if success:
country_str = f" ({country})" if country else ""
logger.info(f"Successfully saved proxy config from API for {provider}{country_str}")
return True, "Proxy configuration saved successfully"
else:
return False, "Failed to save proxy configuration to file"
except Exception as e:
logger.error(f"Error saving proxy config from API for {provider_name}: {e}")
return False, f"Internal error: {str(e)}"
@staticmethod
def _parse_proxy_from_dict(proxy_data: Dict[str, Any]) -> Optional[ProxyConfig]:
"""
Parse proxy dictionary and create ProxyConfig object
Args:
proxy_data: Dictionary with proxy data
Returns:
ProxyConfig instance or None if invalid
"""
from ..models.proxy_models import ProxyConfig, ProxyAuth, ProxyType, ProxyScope
# Required fields
if 'host' not in proxy_data or not proxy_data['host']:
return None
if 'port' not in proxy_data:
return None
try:
port = int(proxy_data['port'])
except (ValueError, TypeError):
return None
# Optional proxy type
proxy_type = ProxyType.HTTP
if 'proxy_type' in proxy_data:
try:
proxy_type = ProxyType(proxy_data['proxy_type'].lower())
except (ValueError, AttributeError):
pass # Use default
# Optional authentication
auth = None
if 'username' in proxy_data and 'password' in proxy_data:
if proxy_data['username'] and proxy_data['password']:
auth = ProxyAuth(
username=proxy_data['username'].strip(),
password=proxy_data['password'].strip()
)
# Optional scope (default to all enabled)
scope = ProxyScope()
if 'scope' in proxy_data and isinstance(proxy_data['scope'], dict):
scope_data = proxy_data['scope']
scope = ProxyScope(
api_calls=scope_data.get('api_calls', True),
authentication=scope_data.get('authentication', True),
manifests=scope_data.get('manifests', True),
license=scope_data.get('license', True),
all=scope_data.get('all', True)
)
# Optional settings
timeout = proxy_data.get('timeout', 30)
verify_ssl = proxy_data.get('verify_ssl', True)
return ProxyConfig(
host=proxy_data['host'].strip(),
port=port,
proxy_type=proxy_type,
auth=auth,
scope=scope,
timeout=timeout,
verify_ssl=verify_ssl
)
def delete_provider_credentials_from_api(self, provider_name: str) -> Tuple[bool, str]:
"""
Delete credentials via API request
Args:
provider_name: Provider name, optionally with country (e.g., "joyn_de")
Returns:
Tuple of (success: bool, message: str)
"""
try:
# Parse provider and country
provider, country = self.parse_provider_country(provider_name)
# Check if provider is registered
if not self.is_provider_registered(provider):
return False, f'Provider "{provider}" is not registered'
# Delete credentials
success = self.credential_manager.delete_credentials(provider, country)
if success:
country_str = f" ({country})" if country else ""
logger.info(f"Successfully deleted credentials from API for {provider}{country_str}")
return True, "Credentials deleted successfully"
else:
return False, "Failed to delete credentials"
except Exception as e:
logger.error(f"Error deleting credentials from API for {provider_name}: {e}")
return False, f"Internal error: {str(e)}"
def delete_provider_proxy_from_api(self, provider_name: str) -> Tuple[bool, str]:
"""
Delete proxy configuration via API request
Args:
provider_name: Provider name, optionally with country (e.g., "joyn_de")
Returns:
Tuple of (success: bool, message: str)
"""
try:
# Parse provider and country
provider, country = self.parse_provider_country(provider_name)
# Check if provider is registered
if not self.is_provider_registered(provider):
return False, f'Provider "{provider}" is not registered'
# Delete proxy config
success = self.proxy_manager.remove_proxy_config(provider, country)
if success:
country_str = f" ({country})" if country else ""
logger.info(f"Successfully deleted proxy config from API for {provider}{country_str}")
return True, "Proxy configuration deleted successfully"
else:
return False, "Failed to delete proxy configuration"
except Exception as e:
logger.error(f"Error deleting proxy config from API for {provider_name}: {e}")
return False, f"Internal error: {str(e)}"
# For imports that expect the old interface
UnifiedSettingsManager = SettingsManager # Backward compatibility alias
lib/streaming_providers/base/utils/environment.py
+307
View File
@@ -0,0 +1,307 @@
# streaming_providers/base/utils/environment.py
"""
Central environment detection and service management.
Provides unified access to VFS, logger, settings bridge, and other services.
"""
import os
import sys
import json
from typing import Dict, Any, Optional, TYPE_CHECKING, Union
from pathlib import Path
# Type hints to avoid circular imports
if TYPE_CHECKING:
from .vfs import VFS
from .logger import BaseLogger # Changed from 'logger' to 'BaseLogger'
from ..settings.kodi_settings_bridge import KodiSettingsBridge
class EnvironmentManager:
"""
Central manager for environment detection and service coordination.
"""
_instance: Optional['EnvironmentManager'] = None
def __new__(cls) -> 'EnvironmentManager':
if cls._instance is None:
cls._instance = super(EnvironmentManager, cls).__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self) -> None:
if self._initialized:
return
self._initialized = True
# Check for Kodi availability
try:
import xbmcaddon
import xbmcvfs
self._is_kodi = True
self._kodi_import_error: Optional[Exception] = None
except ImportError as import_err:
self._is_kodi = False
self._kodi_import_error = import_err
self._addon: Any = None
self._config: Dict[str, Any] = {}
self._services: Dict[str, Any] = {}
# Initialize based on environment
if self._is_kodi:
self._init_kodi()
else:
self._init_standalone()
self._load_config()
def _init_kodi(self) -> None:
"""Initialize Kodi-specific components"""
try:
# Import inside the method where we know Kodi is available
import xbmcaddon as kodi_xbmcaddon
import xbmcvfs as kodi_xbmcvfs
# Create addon instance
self._addon = kodi_xbmcaddon.Addon()
self._config['environment'] = 'kodi'
self._config['addon_id'] = self._addon.getAddonInfo('id')
self._config['addon_name'] = self._addon.getAddonInfo('name')
self._config['addon_version'] = self._addon.getAddonInfo('version')
self._config['addon_path'] = self._addon.getAddonInfo('path')
# Get profile path
profile_info = self._addon.getAddonInfo('profile')
profile_path = kodi_xbmcvfs.translatePath(profile_info)
self._config['profile_path'] = str(profile_path)
# Get settings
default_country = self._addon.getSetting('default_country')
self._config['default_country'] = str(default_country) if default_country else 'DE'
server_port = self._addon.getSetting('server_port')
try:
self._config['server_port'] = int(str(server_port)) if server_port else 7777
except ValueError:
self._config['server_port'] = 7777
except Exception as init_error: # noqa: B902
# Log the error and fallback to standalone
self._log_init_error("Kodi initialization failed", init_error)
self._is_kodi = False
self._init_standalone()
@staticmethod
def _log_init_error(message: str, error: Exception) -> None:
"""Log initialization errors (static method)"""
# We can't use logger here yet, so print to stderr
print(f"{message}: {error}", file=sys.stderr)
def _init_standalone(self) -> None:
"""Initialize standalone mode components"""
self._config['environment'] = 'standalone'
self._config['addon_id'] = 'ultimate-backend-standalone'
self._config['addon_name'] = 'Ultimate Backend'
self._config['addon_version'] = '1.0.0'
self._config['addon_path'] = os.path.dirname(os.path.abspath(__file__))
# Default configuration paths
config_home = os.environ.get('XDG_CONFIG_HOME') or os.path.join(str(Path.home()), '.config')
self._config['config_dir'] = os.path.join(config_home, 'ultimate-backend')
self._config['profile_path'] = self._config['config_dir']
# Load environment variables with defaults
self._config['default_country'] = os.environ.get('DEFAULT_COUNTRY', 'DE')
try:
self._config['server_port'] = int(os.environ.get('SERVER_PORT', '7777'))
except ValueError as port_error:
print(f"Invalid server port, using default: {port_error}", file=sys.stderr)
self._config['server_port'] = 7777
# Ensure config directory exists
try:
os.makedirs(self._config['config_dir'], exist_ok=True)
except OSError as dir_error:
print(f"Failed to create config directory: {dir_error}", file=sys.stderr)
# Use temp directory as fallback
import tempfile
self._config['config_dir'] = tempfile.mkdtemp(prefix='ultimate-backend-')
self._config['profile_path'] = self._config['config_dir']
def _load_config(self) -> None:
"""Load additional configuration from files"""
config_file = os.path.join(self._config['profile_path'], 'config.json')
if os.path.exists(config_file):
try:
with open(config_file, 'r', encoding='utf-8') as f:
file_config = json.load(f)
# Update config with file contents
for key, value in file_config.items():
if isinstance(value, (str, int, float, bool, type(None))):
self._config[key] = value
except json.JSONDecodeError as json_error:
print(f"Invalid JSON in config file: {json_error}", file=sys.stderr)
except OSError as io_error:
print(f"Failed to read config file: {io_error}", file=sys.stderr)
def is_kodi(self) -> bool:
"""Check if running in Kodi environment"""
return self._is_kodi
def get_environment(self) -> str:
"""Get current environment name"""
return str(self._config.get('environment', 'unknown'))
def get_config(self, key: str, default: Any = None) -> Any:
"""Get configuration value"""
return self._config.get(key, default)
def set_config(self, key: str, value: Union[str, int, float, bool, None]) -> None:
"""Set configuration value"""
self._config[key] = value
# Auto-save to config file in standalone mode
if not self._is_kodi:
config_file = os.path.join(self._config['profile_path'], 'config.json')
try:
# Read existing config
existing_config = {}
if os.path.exists(config_file):
with open(config_file, 'r', encoding='utf-8') as f:
existing_config = json.load(f)
# Update with new value
existing_config[key] = value
# Write back
with open(config_file, 'w', encoding='utf-8') as f:
json.dump(existing_config, f, indent=2, ensure_ascii=False)
except OSError as save_error:
print(f"Failed to save config: {save_error}", file=sys.stderr)
except (TypeError, ValueError) as type_error:
print(f"Config contains non-serializable data: {type_error}", file=sys.stderr)
def get_vfs(self, subdir: str = "", config_dir: Optional[str] = None) -> 'VFS':
"""Get VFS instance with appropriate configuration"""
# Import here to avoid circular imports
from .vfs import VFS
# Use provided config_dir, or environment default
if config_dir is None:
config_dir = self._config.get('profile_path', '')
return VFS(config_dir=config_dir, addon_subdir=subdir)
@staticmethod
def get_logger() -> 'BaseLogger':
"""Get logger instance for current environment (static method)"""
# Import here to avoid circular imports
from .logger import logger as logger_instance
return logger_instance
def get_settings_bridge(self, addon_id: Optional[str] = None,
config_dir: Optional[str] = None) -> 'KodiSettingsBridge':
"""Get settings bridge instance"""
# Import here to avoid circular imports
from ..settings.kodi_settings_bridge import KodiSettingsBridge
if config_dir is None:
config_dir = self._config.get('profile_path', '')
return KodiSettingsBridge(addon_id=addon_id, config_dir=config_dir)
def get_manager(self) -> Any:
"""Get the configured streaming provider manager"""
try:
# Lazy import to avoid circular dependencies
import importlib
# Adjust this import path based on your actual module structure
module = importlib.import_module('streaming_providers')
if hasattr(module, 'get_configured_manager'):
manager_func = module.get_configured_manager
if callable(manager_func):
return manager_func()
else:
raise ImportError("get_configured_manager is not callable")
else:
raise ImportError("get_configured_manager not found in streaming_providers module")
except ImportError as manager_error:
# Log error using the logger once we have it
logger_instance = self.get_logger()
logger_instance.error(f"Failed to import manager: {manager_error}")
raise
def get_service_config(self) -> Dict[str, Any]:
"""Get configuration for running the service"""
return {
'is_kodi': self._is_kodi,
'port': self._config.get('server_port', 7777),
'default_country': self._config.get('default_country', 'DE'),
'profile_path': self._config.get('profile_path', ''),
'addon_path': self._config.get('addon_path', '')
}
def debug_info(self) -> Dict[str, Any]:
"""Get debug information about the environment"""
info: Dict[str, Any] = {
'environment': self.get_environment(),
'is_kodi': self._is_kodi,
'python_version': sys.version,
'platform': sys.platform,
}
# Add Kodi-specific info if available
if self._is_kodi and self._addon:
info['kodi_addon_id'] = self._addon.getAddonInfo('id')
info['kodi_addon_version'] = self._addon.getAddonInfo('version')
# Add import error info if present
if hasattr(self, '_kodi_import_error') and self._kodi_import_error:
info['kodi_import_error'] = str(self._kodi_import_error)
# Create a safe config summary without sensitive paths
safe_config: Dict[str, Any] = {}
for key, value in self._config.items():
if not key.endswith('_path') and key != 'config_dir' and key not in ['profile_path', 'addon_path']:
if isinstance(value, (str, int, float, bool, type(None))):
safe_config[key] = value
else:
safe_config[key] = str(type(value))
info['config_summary'] = safe_config
return info
# Global singleton instance
_env_manager: Optional[EnvironmentManager] = None
def get_environment_manager() -> EnvironmentManager:
"""Get the global environment manager instance"""
global _env_manager
if _env_manager is None:
_env_manager = EnvironmentManager()
return _env_manager
# Convenience functions for common operations
def is_kodi_environment() -> bool:
"""Check if we're running in Kodi environment (convenience function)"""
return get_environment_manager().is_kodi()
def get_logger_instance() -> 'BaseLogger':
"""Get logger instance (convenience function)"""
return EnvironmentManager.get_logger()
def get_vfs_instance(subdir: str = "", config_dir: Optional[str] = None) -> 'VFS':
"""Get VFS instance (convenience function)"""
return get_environment_manager().get_vfs(subdir, config_dir)
lib/streaming_providers/base/utils/logger.py
+141 -39
View File
@@ -1,63 +1,165 @@
# streaming_providers/base/utils/logger.py
import xbmc
"""
Centralized logging module for Ultimate Backend.
Provides environment-aware logging for both Kodi and standalone modes.
"""
import sys
import logging
# Import environment manager
from .environment import get_environment_manager
# Get environment manager
_env_manager_instance = get_environment_manager()
class XBMCLogger:
"""Centralized logging using XBMC's logging system."""
class BaseLogger:
"""Base logger interface that all logger implementations must follow"""
def __init__(self, addon_name: str, addon_version: str):
self.addon_name = addon_name
self.addon_version = addon_version
self.prefix = f"[{addon_name} v{addon_version}]"
def log(self, message: str, level: int = xbmc.LOGINFO) -> None:
"""Main logging method.
Args:
message: The message to log
level: One of xbmc.LOGDEBUG, LOGINFO, LOGWARNING, LOGERROR, LOGFATAL
"""
xbmc.log(f"{self.prefix} {message}", level)
def __init__(self, logger_name: str, logger_version: str):
self.logger_name = logger_name
self.logger_version = logger_version
self.prefix = f"[{logger_name} v{logger_version}]"
def debug(self, message: str) -> None:
self.log(message, xbmc.LOGDEBUG)
"""Log debug message"""
raise NotImplementedError
def info(self, message: str) -> None:
self.log(message, xbmc.LOGINFO)
"""Log info message"""
raise NotImplementedError
def warning(self, message: str) -> None:
self.log(message, xbmc.LOGWARNING)
"""Log warning message"""
raise NotImplementedError
def error(self, message: str, exc_info=False) -> None:
self.log(message, xbmc.LOGERROR)
def error(self, message: str, exc_info: bool = False) -> None:
"""Log error message"""
raise NotImplementedError
def critical(self, message: str) -> None:
self.log(message, xbmc.LOGFATAL)
"""Log critical message"""
raise NotImplementedError
# Specialized methods
def log_auth_event(self, provider: str, event: str, details: str = "") -> None:
"""Specialized logging method for authentication events."""
message = f"AUTH [{provider}] {event}"
"""Log authentication event"""
log_message = f"AUTH [{provider}] {event}"
if details:
message += f" - {details}"
self.info(message)
log_message += f" - {details}"
self.info(log_message)
def log_credential_event(self, provider: str, event: str, details: str = "") -> None:
"""Specialized logging method for credential management events."""
message = f"CRED [{provider}] {event}"
"""Log credential event"""
log_message = f"CRED [{provider}] {event}"
if details:
message += f" - {details}"
self.info(message)
log_message += f" - {details}"
self.info(log_message)
def log_session_event(self, provider: str, event: str, details: str = "") -> None:
"""Specialized logging method for session management events."""
message = f"SESSION [{provider}] {event}"
"""Log session event"""
log_message = f"SESSION [{provider}] {event}"
if details:
message += f" - {details}"
self.debug(message)
log_message += f" - {details}"
self.debug(log_message)
# Initialize with your addon info
logger = XBMCLogger(
addon_name="Ultimate Backend",
addon_version="1.0.0" # You could get this from your addon.xml
)
def create_logger() -> BaseLogger:
"""Create appropriate logger instance based on environment"""
# Get configuration
app_name = _env_manager_instance.get_config('addon_name', 'Ultimate Backend')
app_version = _env_manager_instance.get_config('addon_version', '1.0.0')
if _env_manager_instance.is_kodi():
# Try to create Kodi logger
try:
import xbmc
class XBMCLogger(BaseLogger):
"""Centralized logging using XBMC's logging system."""
def debug(self, message: str) -> None:
xbmc.log(f"{self.prefix} {message}", xbmc.LOGDEBUG)
def info(self, message: str) -> None:
xbmc.log(f"{self.prefix} {message}", xbmc.LOGINFO)
def warning(self, message: str) -> None:
xbmc.log(f"{self.prefix} {message}", xbmc.LOGWARNING)
def error(self, message: str, exc_info: bool = False) -> None:
xbmc.log(f"{self.prefix} {message}", xbmc.LOGERROR)
def critical(self, message: str) -> None:
xbmc.log(f"{self.prefix} {message}", xbmc.LOGFATAL)
return XBMCLogger(str(app_name), str(app_version))
except ImportError as xbmc_import_error:
print(f"Failed to import xbmc: {xbmc_import_error}", file=sys.stderr)
# Fall through to standard logger
# Create standard logger (fallback for non-Kodi or failed Kodi)
class StandardLogger(BaseLogger):
"""Standard Python logging for non-Kodi environments."""
def __init__(self, logger_name: str, logger_version: str):
super().__init__(logger_name, logger_version)
# Configure logging
self._logger = logging.getLogger(logger_name)
# Only add handlers if none exist
if not self._logger.handlers:
# Console handler
console_handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter(
f'%(asctime)s {self.prefix} %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console_handler.setFormatter(formatter)
self._logger.addHandler(console_handler)
# File handler (optional)
log_dir = _env_manager_instance.get_config('profile_path')
if log_dir:
import os
log_file = os.path.join(str(log_dir), 'ultimate-backend.log')
try:
file_handler = logging.FileHandler(log_file, encoding='utf-8')
file_handler.setFormatter(formatter)
self._logger.addHandler(file_handler)
except (OSError, PermissionError) as file_handler_error:
# Log to console only if file logging fails
print(f"Failed to create file handler: {file_handler_error}", file=sys.stderr)
self._logger.setLevel(logging.DEBUG)
def debug(self, message: str) -> None:
self._logger.debug(message)
def info(self, message: str) -> None:
self._logger.info(message)
def warning(self, message: str) -> None:
self._logger.warning(message)
def error(self, message: str, exc_info: bool = False) -> None:
if exc_info:
self._logger.error(message, exc_info=True)
else:
self._logger.error(message)
def critical(self, message: str) -> None:
self._logger.critical(message)
return StandardLogger(str(app_name), str(app_version))
# Create global logger instance - this is the actual logger object users will import
logger: BaseLogger = create_logger()
# Export the BaseLogger type for type hints
__all__ = ['BaseLogger', 'logger']
lib/streaming_providers/base/utils/mpd_cache.py
+6 -3
View File
@@ -15,15 +15,18 @@ class MPDCacheManager:
self.vfs = VFS(addon_subdir="mpd_cache")
logger.debug(f"MPD cache initialized at: {self.vfs.base_path}")
def _get_cache_key(self, provider: str, channel_id: str) -> str:
@staticmethod
def _get_cache_key(provider: str, channel_id: str) -> str:
"""Generate cache key for provider/channel"""
return f"{provider}_{channel_id}"
def _get_manifest_filename(self, cache_key: str) -> str:
@staticmethod
def _get_manifest_filename(cache_key: str) -> str:
"""Get filename for cached manifest"""
return f"{cache_key}.xml"
def _get_meta_filename(self, cache_key: str) -> str:
@staticmethod
def _get_meta_filename(cache_key: str) -> str:
"""Get filename for cache metadata"""
return f"{cache_key}.meta"
lib/streaming_providers/base/utils/vfs.py
+104 -85
View File
@@ -6,24 +6,14 @@ Provides transparent file operations for both Kodi and regular Python environmen
import json
import os
from typing import Optional, Any, List
from pathlib import Path
from typing import Optional, Any, List, Dict, Tuple
# Import centralized environment manager
from .environment import get_environment_manager, is_kodi_environment
# Import centralized logger
from .logger import logger
# Kodi imports - with fallback for non-Kodi environments
try:
import xbmc
import xbmcvfs
import xbmcaddon
KODI_AVAILABLE = True
logger.info("Kodi VFS environment detected")
except ImportError:
KODI_AVAILABLE = False
logger.info("Standard filesystem environment detected")
class VFS:
"""
@@ -44,6 +34,8 @@ class VFS:
self.addon_subdir = addon_subdir
self._base_path = None
self._explicit_config_dir = config_dir
self._env_manager = get_environment_manager()
logger.debug(f"VFS initialized with config_dir={config_dir}, addon_subdir={addon_subdir}")
@property
@@ -54,31 +46,20 @@ class VFS:
# Use explicitly provided config directory
self._base_path = self._explicit_config_dir
logger.info(f"Using explicit config directory: {self._base_path}")
elif KODI_AVAILABLE:
# Use Kodi's addon data directory
try:
addon = xbmcaddon.Addon()
# Use xbmcvfs.translatePath instead of xbmc.translatePath for Kodi 19+
addon_profile = xbmcvfs.translatePath(addon.getAddonInfo('profile'))
if self.addon_subdir:
self._base_path = os.path.join(addon_profile, self.addon_subdir).replace('\\', '/')
else:
self._base_path = addon_profile.replace('\\', '/')
logger.info(f"Kodi base path: {self._base_path}")
except Exception as e:
logger.error(f"Error getting Kodi addon path: {e}")
# Fallback to temp directory
try:
self._base_path = xbmcvfs.translatePath("special://temp/streaming_providers")
except:
self._base_path = "/tmp/streaming_providers"
else:
# Use standard filesystem
# Use profile path from environment manager
profile_path = self._env_manager.get_config('profile_path', '')
if self.addon_subdir:
self._base_path = str(Path.home() / '.streaming_providers' / self.addon_subdir)
if is_kodi_environment():
# Kodi uses forward slashes
self._base_path = os.path.join(profile_path, self.addon_subdir).replace('\\', '/')
else:
# Standard filesystem
self._base_path = os.path.join(profile_path, self.addon_subdir)
else:
self._base_path = str(Path.home() / '.streaming_providers')
logger.info(f"Standard filesystem base path: {self._base_path}")
self._base_path = profile_path
logger.info(f"Base path from environment: {self._base_path}")
# Ensure base directory exists
self.mkdirs('')
@@ -95,7 +76,7 @@ class VFS:
Returns:
Joined path string
"""
if KODI_AVAILABLE:
if is_kodi_environment():
# Kodi VFS uses forward slashes
path = self.base_path
for part in parts:
@@ -103,12 +84,8 @@ class VFS:
path = path.rstrip('/') + '/' + str(part).lstrip('/')
return path
else:
# Use pathlib for standard filesystem
path = Path(self.base_path)
for part in parts:
if part:
path = path / str(part)
return str(path)
# Use os.path.join for standard filesystem
return os.path.join(self.base_path, *[str(p) for p in parts if p])
def exists(self, filepath: str) -> bool:
"""
@@ -124,10 +101,12 @@ class VFS:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
return xbmcvfs.exists(filepath)
else:
return Path(filepath).exists()
import pathlib
return pathlib.Path(filepath).exists()
except Exception as e:
logger.error(f"Error checking if {filepath} exists: {e}")
return False
@@ -146,14 +125,16 @@ class VFS:
if not os.path.isabs(dirpath):
dirpath = self.join_path(dirpath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
if not xbmcvfs.exists(dirpath):
result = xbmcvfs.mkdirs(dirpath)
logger.debug(f"Kodi mkdirs {dirpath}: {result}")
return result
return True
else:
Path(dirpath).mkdir(parents=True, exist_ok=True)
import pathlib
pathlib.Path(dirpath).mkdir(parents=True, exist_ok=True)
return True
except Exception as e:
logger.error(f"Error creating directory {dirpath}: {e}")
@@ -174,14 +155,16 @@ class VFS:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
if not xbmcvfs.exists(filepath):
return None
with xbmcvfs.File(filepath, 'r') as f:
content = f.read()
return content if content else None
else:
path = Path(filepath)
import pathlib
path = pathlib.Path(filepath)
if not path.exists():
return None
with open(path, 'r', encoding=encoding) as f:
@@ -206,7 +189,8 @@ class VFS:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
# Ensure directory exists
dir_path = '/'.join(filepath.split('/')[:-1])
if dir_path and not xbmcvfs.exists(dir_path):
@@ -217,7 +201,8 @@ class VFS:
logger.debug(f"Kodi file write: {bytes_written} bytes to {filepath}")
return bytes_written > 0
else:
path = Path(filepath)
import pathlib
path = pathlib.Path(filepath)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'w', encoding=encoding) as f:
f.write(content)
@@ -240,12 +225,14 @@ class VFS:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
if xbmcvfs.exists(filepath):
return xbmcvfs.delete(filepath)
return True
else:
path = Path(filepath)
import pathlib
path = pathlib.Path(filepath)
if path.exists():
path.unlink()
return True
@@ -311,7 +298,9 @@ class VFS:
elif not os.path.isabs(dirpath):
dirpath = self.join_path(dirpath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
import fnmatch
if not xbmcvfs.exists(dirpath):
return []
@@ -321,10 +310,10 @@ class VFS:
return files
else:
# Simple pattern matching
import fnmatch
return [f for f in files if fnmatch.fnmatch(f, pattern)]
else:
path = Path(dirpath)
import pathlib
path = pathlib.Path(dirpath)
if not path.exists() or not path.is_dir():
return []
@@ -350,13 +339,15 @@ class VFS:
if not os.path.isabs(filepath):
filepath = self.join_path(filepath)
if KODI_AVAILABLE:
if is_kodi_environment():
import xbmcvfs
if not xbmcvfs.exists(filepath):
return None
stat = xbmcvfs.Stat(filepath)
return stat.st_size()
else:
path = Path(filepath)
import pathlib
path = pathlib.Path(filepath)
if not path.exists():
return None
return path.stat().st_size
@@ -392,15 +383,18 @@ class VFS:
Dictionary with debug information
"""
info = {
'kodi_available': KODI_AVAILABLE,
'kodi_available': is_kodi_environment(),
'base_path': self.base_path,
'base_path_exists': self.exists(''),
'explicit_config_dir': self._explicit_config_dir,
'addon_subdir': self.addon_subdir
'addon_subdir': self.addon_subdir,
'environment': self._env_manager.get_environment()
}
if KODI_AVAILABLE:
if is_kodi_environment():
try:
import xbmcaddon
import xbmc
addon = xbmcaddon.Addon()
info.update({
'addon_id': addon.getAddonInfo('id'),
@@ -413,13 +407,15 @@ class VFS:
return info
# Convenience functions for global VFS instance
_global_vfs = None
# Cache for VFS instances with different configurations
_vfs_cache: Dict[Tuple[Optional[str], str], 'VFS'] = {}
def get_vfs(config_dir: Optional[str] = None, addon_subdir: str = "") -> VFS:
def get_vfs(config_dir: Optional[str] = None, addon_subdir: str = "") -> 'VFS':
"""
Get global VFS instance
Get VFS instance for specific configuration
Uses a cache to avoid creating multiple instances with same configuration
Args:
config_dir: Optional explicit config directory
@@ -428,49 +424,72 @@ def get_vfs(config_dir: Optional[str] = None, addon_subdir: str = "") -> VFS:
Returns:
VFS instance
"""
global _vfs_cache
cache_key = (config_dir, addon_subdir)
if cache_key not in _vfs_cache:
_vfs_cache[cache_key] = VFS(config_dir, addon_subdir)
return _vfs_cache[cache_key]
# For backward compatibility with existing code
_global_vfs = None
def get_global_vfs(config_dir: Optional[str] = None, addon_subdir: str = "") -> 'VFS':
"""
Get global VFS instance (for backward compatibility)
Note: Consider using get_vfs() for new code
"""
global _global_vfs
if _global_vfs is None or _global_vfs._explicit_config_dir != config_dir or _global_vfs.addon_subdir != addon_subdir:
_global_vfs = VFS(config_dir, addon_subdir)
return _global_vfs
vfs = get_vfs(config_dir, addon_subdir)
_global_vfs = vfs # Keep reference for backward compatibility
return vfs
# Convenience functions that use global VFS
def exists(filepath: str, config_dir: Optional[str] = None) -> bool:
# Convenience functions that use VFS cache
def exists(filepath: str, config_dir: Optional[str] = None, addon_subdir: str = "") -> bool:
"""Check if file exists"""
return get_vfs(config_dir).exists(filepath)
return get_vfs(config_dir, addon_subdir).exists(filepath)
def mkdirs(dirpath: str, config_dir: Optional[str] = None) -> bool:
def mkdirs(dirpath: str, config_dir: Optional[str] = None, addon_subdir: str = "") -> bool:
"""Create directories"""
return get_vfs(config_dir).mkdirs(dirpath)
return get_vfs(config_dir, addon_subdir).mkdirs(dirpath)
def read_text(filepath: str, encoding: str = 'utf-8', config_dir: Optional[str] = None) -> Optional[str]:
def read_text(filepath: str, encoding: str = 'utf-8', config_dir: Optional[str] = None, addon_subdir: str = "") -> \
Optional[str]:
"""Read text file"""
return get_vfs(config_dir).read_text(filepath, encoding)
return get_vfs(config_dir, addon_subdir).read_text(filepath, encoding)
def write_text(filepath: str, content: str, encoding: str = 'utf-8', config_dir: Optional[str] = None) -> bool:
def write_text(filepath: str, content: str, encoding: str = 'utf-8', config_dir: Optional[str] = None,
addon_subdir: str = "") -> bool:
"""Write text file"""
return get_vfs(config_dir).write_text(filepath, content, encoding)
return get_vfs(config_dir, addon_subdir).write_text(filepath, content, encoding)
def read_json(filepath: str, config_dir: Optional[str] = None) -> Optional[dict]:
def read_json(filepath: str, config_dir: Optional[str] = None, addon_subdir: str = "") -> Optional[dict]:
"""Read JSON file"""
return get_vfs(config_dir).read_json(filepath)
return get_vfs(config_dir, addon_subdir).read_json(filepath)
def write_json(filepath: str, data: Any, indent: int = 2, config_dir: Optional[str] = None) -> bool:
def write_json(filepath: str, data: Any, indent: int = 2, config_dir: Optional[str] = None,
addon_subdir: str = "") -> bool:
"""Write JSON file"""
return get_vfs(config_dir).write_json(filepath, data, indent)
return get_vfs(config_dir, addon_subdir).write_json(filepath, data, indent)
def delete(filepath: str, config_dir: Optional[str] = None) -> bool:
def delete(filepath: str, config_dir: Optional[str] = None, addon_subdir: str = "") -> bool:
"""Delete file"""
return get_vfs(config_dir).delete(filepath)
return get_vfs(config_dir, addon_subdir).delete(filepath)
def join_path(*parts, config_dir: Optional[str] = None) -> str:
def join_path(*parts, config_dir: Optional[str] = None, addon_subdir: str = "") -> str:
"""Join path components"""
vfs = get_vfs(config_dir)
vfs = get_vfs(config_dir, addon_subdir)
return vfs.join_path(*parts)
requirements.txt
View File
service.py
+370 -30
View File
@@ -3,30 +3,44 @@ import os
import sys
import threading
from datetime import datetime
import xbmc
import xbmcaddon
import time
import json
from bottle import Bottle, run, request, response, redirect, HTTPResponse
from urllib.parse import urlencode, parse_qsl
# Get addon settings
ADDON = xbmcaddon.Addon()
ADDON_PATH = ADDON.getAddonInfo('path')
LIB_PATH = os.path.join(ADDON_PATH, 'lib')
sys.path.insert(0, LIB_PATH)
# Add lib path for imports
script_dir = os.path.dirname(os.path.abspath(__file__))
LIB_PATH = os.path.join(script_dir, 'lib')
if os.path.exists(LIB_PATH):
sys.path.insert(0, LIB_PATH)
try:
from streaming_providers import get_configured_manager
from streaming_providers.base.models import StreamingChannel
from streaming_providers.base.utils import logger, VFS, MPDRewriter, MPDCacheManager
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
except ImportError as import_err:
xbmc.log(f"Ultimate Backend: Critical import failed - {str(import_err)}", xbmc.LOGERROR)
print(f"Ultimate Backend: Critical import failed - {str(import_err)}", file=sys.stderr)
raise
class UltimateService:
def __init__(self):
def __init__(self, config_dir: str = None):
self.app = Bottle()
# Get environment manager
self.env_manager = get_environment_manager()
# Override config directory if provided
if config_dir:
self.env_manager.set_config('profile_path', config_dir)
# Get settings
self.server_port = self.env_manager.get_config('server_port', 7777)
self.default_country = self.env_manager.get_config('default_country', 'DE')
# Initialize manager
try:
self.manager = get_configured_manager()
logger.info("Manager initialized successfully")
@@ -35,7 +49,7 @@ class UltimateService:
raise
# Initialize VFS for M3U caching
self.vfs = VFS(addon_subdir="m3u_cache")
self.vfs = get_vfs_instance(subdir="m3u_cache")
logger.info(f"VFS initialized for M3U caching: {self.vfs.base_path}")
self.mpd_cache = MPDCacheManager()
@@ -43,6 +57,21 @@ class UltimateService:
self.setup_routes()
def _get_setting(self, setting_id: str, default: str = None) -> str:
"""Get setting value from appropriate source"""
# Try Kodi settings first if in Kodi environment
if is_kodi_environment():
try:
import xbmcaddon
addon = xbmcaddon.Addon()
value = addon.getSetting(setting_id)
return value if value else default
except Exception as e:
logger.debug(f"Could not get Kodi setting {setting_id}: {e}")
# Fallback to environment manager config
return self.env_manager.get_config(setting_id, default)
def _get_proxied_manifest(self, provider: str, channel_id: str) -> str:
"""
Get proxied and rewritten MPD manifest for a channel.
@@ -300,7 +329,8 @@ class UltimateService:
return directives
def _process_license_headers(self, req_headers):
@staticmethod
def _process_license_headers(req_headers):
"""
Process license headers and convert to URL-encoded format.
@@ -441,7 +471,7 @@ class UltimateService:
def list_providers():
try:
provider_names = self.manager.list_providers()
default_country = ADDON.getSetting('default_country') or 'DE'
default_country = self._get_setting('default_country', 'DE') # Changed
# Enhanced response with provider details including labels
providers_details = []
@@ -460,7 +490,7 @@ class UltimateService:
})
return {
'providers': providers_details, # Now includes labels and countries
'providers': providers_details,
'default_country': default_country
}
except Exception as api_err:
@@ -1082,35 +1112,345 @@ class UltimateService:
response.status = 500
return {'error': str(e)}
def run_service():
service = UltimateService()
port = int(ADDON.getSetting("server_port") or 7777)
@self.app.route('/api/providers/<provider>/auth/status')
def get_provider_auth_status(provider):
"""
Get authentication status for a provider
Returns current authentication state including:
- Authentication state (not_authenticated, credentials_only, user_authenticated, client_authenticated)
- Credential type
- Token validity and expiration
- Whether provider is ready to use
Example: GET /api/providers/joyn_de/auth/status
"""
try:
status = self.manager.settings_manager.get_auth_status(provider)
response.content_type = 'application/json; charset=utf-8'
return status
except Exception as api_err:
logger.error(f"API Error in /api/providers/{provider}/auth/status: {str(api_err)}")
response.status = 500
return {'error': f'Internal server error: {str(api_err)}'}
@self.app.route('/api/providers/<provider>/credentials', method='POST')
def save_provider_credentials(provider):
"""
Save credentials for a provider via API
Accepts JSON body with credentials:
- User/password: {"username": "...", "password": "...", "client_id": "..." (optional)}
- Client credentials: {"client_id": "...", "client_secret": "..."}
Example: POST /api/providers/joyn_de/credentials
Body: {"username": "user@example.com", "password": "secret123"}
"""
try:
# Parse JSON body
try:
credentials_data = request.json
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {'error': 'Invalid JSON in request body'}
if not credentials_data:
response.status = 400
return {'error': 'Request body must contain credentials data'}
# Validate it's a dictionary
if not isinstance(credentials_data, dict):
response.status = 400
return {'error': 'Credentials data must be a JSON object'}
# Save credentials
success, message = self.manager.settings_manager.save_provider_credentials_from_api(
provider, credentials_data
)
if success:
response.status = 200
response.content_type = 'application/json; charset=utf-8'
return {
'success': True,
'provider': provider,
'message': message
}
else:
# Determine appropriate status code
if 'not registered' in message.lower():
response.status = 404
elif 'invalid' in message.lower() or 'validation failed' in message.lower():
response.status = 400
else:
response.status = 500
return {'error': message}
except Exception as api_err:
logger.error(f"API Error in POST /api/providers/{provider}/credentials: {str(api_err)}")
response.status = 500
return {'error': f'Internal server error: {str(api_err)}'}
@self.app.route('/api/providers/<provider>/credentials', method='DELETE')
def delete_provider_credentials(provider):
"""
Delete credentials for a provider via API
Example: DELETE /api/providers/joyn_de/credentials
"""
try:
success, message = self.manager.settings_manager.delete_provider_credentials_from_api(provider)
if success:
response.status = 200
response.content_type = 'application/json; charset=utf-8'
return {
'success': True,
'provider': provider,
'message': message
}
else:
# Determine appropriate status code
if 'not registered' in message.lower():
response.status = 404
else:
response.status = 500
return {'error': message}
except Exception as api_err:
logger.error(f"API Error in DELETE /api/providers/{provider}/credentials: {str(api_err)}")
response.status = 500
return {'error': f'Internal server error: {str(api_err)}'}
@self.app.route('/api/providers/<provider>/proxy', method='POST')
def save_provider_proxy(provider):
"""
Save proxy configuration for a provider via API
Accepts JSON body with proxy configuration:
Required: {"host": "proxy.example.com", "port": 8080}
Optional: {
"proxy_type": "http", # http, https, socks4, socks5
"username": "proxyuser",
"password": "proxypass",
"timeout": 30,
"verify_ssl": true,
"scope": {
"api_calls": true,
"authentication": true,
"manifests": true,
"license": true,
"all": true
}
}
Example: POST /api/providers/joyn_de/proxy
Body: {"host": "proxy.example.com", "port": 8080}
"""
try:
# Parse JSON body
try:
proxy_data = request.json
except Exception as json_err:
logger.error(f"Invalid JSON in request body: {json_err}")
response.status = 400
return {'error': 'Invalid JSON in request body'}
if not proxy_data:
response.status = 400
return {'error': 'Request body must contain proxy configuration'}
# Validate it's a dictionary
if not isinstance(proxy_data, dict):
response.status = 400
return {'error': 'Proxy data must be a JSON object'}
# Save proxy configuration
success, message = self.manager.settings_manager.save_provider_proxy_from_api(
provider, proxy_data
)
if success:
response.status = 200
response.content_type = 'application/json; charset=utf-8'
return {
'success': True,
'provider': provider,
'message': message
}
else:
# Determine appropriate status code
if 'not registered' in message.lower():
response.status = 404
elif 'invalid' in message.lower() or 'validation failed' in message.lower():
response.status = 400
else:
response.status = 500
return {'error': message}
except Exception as api_err:
logger.error(f"API Error in POST /api/providers/{provider}/proxy: {str(api_err)}")
response.status = 500
return {'error': f'Internal server error: {str(api_err)}'}
@self.app.route('/api/providers/<provider>/proxy', method='DELETE')
def delete_provider_proxy(provider):
"""
Delete proxy configuration for a provider via API
Example: DELETE /api/providers/joyn_de/proxy
"""
try:
success, message = self.manager.settings_manager.delete_provider_proxy_from_api(provider)
if success:
response.status = 200
response.content_type = 'application/json; charset=utf-8'
return {
'success': True,
'provider': provider,
'message': message
}
else:
# Determine appropriate status code
if 'not registered' in message.lower():
response.status = 404
else:
response.status = 500
return {'error': message}
except Exception as api_err:
logger.error(f"API Error in DELETE /api/providers/{provider}/proxy: {str(api_err)}")
response.status = 500
return {'error': f'Internal server error: {str(api_err)}'}
def start_service(service_instance):
"""Start the Bottle server"""
port = service_instance.server_port
logger.info(f"Starting server on port {port}")
run(service.app, host='0.0.0.0', port=port, quiet=True, debug=True)
# Determine if we should run in debug mode
debug_mode = service_instance.env_manager.get_config('debug_mode', False)
run(service_instance.app, host='0.0.0.0', port=port, quiet=not debug_mode, debug=debug_mode)
if __name__ == '__main__':
logger.info("Starting service...")
# Give Kodi time to initialize
import time
time.sleep(5)
def run_kodi_service():
"""Run service within Kodi addon context"""
logger.info("Starting Ultimate Backend service in Kodi mode")
try:
import xbmc
import xbmcaddon
except ImportError:
logger.error("Kodi modules not available!")
print("ERROR: Cannot run in Kodi mode - xbmc/xbmcaddon not available")
return
# Give Kodi time to initialize
time.sleep(3)
try:
# Create service instance
service = UltimateService()
# Start service in background thread
service_thread = threading.Thread(
target=run_service,
target=start_service,
args=(service,),
name="UltimateBackendService"
)
service_thread.daemon = True
service_thread.start()
# Monitor for Kodi shutdown
monitor = xbmc.Monitor()
while not monitor.abortRequested():
if monitor.waitForAbort(5):
break
logger.info("Service stopped")
except Exception as startup_err:
logger.error(f"Failed to start - {str(startup_err)}")
raise
logger.info("Service stopped (Kodi shutdown)")
except Exception as e:
logger.error(f"Failed to start Kodi service: {e}")
raise
def run_standalone_service(config_dir: str = None):
"""Run service in standalone mode"""
logger.info("Starting Ultimate Backend service in standalone mode")
# Create service instance
service = UltimateService(config_dir=config_dir)
# Print startup information
print("=" * 60)
print("Ultimate Backend Streaming Service")
print("=" * 60)
print(f"Mode: Standalone")
print(f"Port: {service.server_port}")
print(f"Default Country: {service.default_country}")
print(f"Config Directory: {service.vfs.base_path}")
print(f"Log Directory: {service.env_manager.get_config('profile_path', 'N/A')}")
print("=" * 60)
print(f"API Endpoints:")
print(f" http://localhost:{service.server_port}/api/providers")
print(f" http://localhost:{service.server_port}/api/m3u")
print(f" http://localhost:{service.server_port}/api/providers/<provider>/m3u")
print("=" * 60)
print("Press Ctrl+C to stop the service")
print("=" * 60)
try:
start_service(service)
except KeyboardInterrupt:
print("\nService stopped by user")
except Exception as e:
print(f"Error running service: {e}")
sys.exit(1)
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='Ultimate Backend Streaming Service')
parser.add_argument('--port', type=int, help='Server port (overrides config)')
parser.add_argument('--config-dir', help='Configuration directory')
parser.add_argument('--debug', action='store_true', help='Enable debug mode')
parser.add_argument('--kodi', action='store_true', help='Force Kodi mode (requires Kodi modules)')
parser.add_argument('--standalone', action='store_true', help='Force standalone mode')
args = parser.parse_args()
# Get environment manager
env_manager = get_environment_manager()
# Apply CLI overrides
if args.port:
env_manager.set_config('server_port', args.port)
logger.info(f"Port overridden via CLI: {args.port}")
if args.debug:
env_manager.set_config('debug_mode', True)
logger.info("Debug mode enabled via CLI")
# Determine execution mode
if args.kodi:
logger.info("Kodi mode forced by CLI argument")
run_kodi_service()
elif args.standalone:
logger.info("Standalone mode forced by CLI argument")
run_standalone_service(config_dir=args.config_dir)
elif is_kodi_environment():
logger.info("Kodi environment detected, running in Kodi mode")
run_kodi_service()
else:
logger.info("Running in standalone mode (default)")
run_standalone_service(config_dir=args.config_dir)