From c1e30fa1c68e368c55a78275b36b2a233752c25c Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 18 Sep 2026 06:50:55 +0200 Subject: [PATCH] split provider --- lib/streaming_providers/base/auth_type.py | 21 + lib/streaming_providers/base/provider.py | 1768 +---------------- .../base/provider_mixins/auth.py | 483 +++++ .../base/provider_mixins/bookmarks.py | 204 ++ .../base/provider_mixins/catchup.py | 284 +++ .../base/provider_mixins/epg.py | 83 + .../base/provider_mixins/favorites.py | 65 + .../base/provider_mixins/http.py | 152 ++ .../base/provider_mixins/metadata.py | 170 ++ .../base/provider_mixins/recordings.py | 61 + .../base/provider_mixins/subscriptions.py | 111 ++ .../base/provider_mixins/timers.py | 116 ++ .../base/provider_mixins/vod.py | 56 + 13 files changed, 1843 insertions(+), 1731 deletions(-) create mode 100644 lib/streaming_providers/base/auth_type.py create mode 100644 lib/streaming_providers/base/provider_mixins/auth.py create mode 100644 lib/streaming_providers/base/provider_mixins/bookmarks.py create mode 100644 lib/streaming_providers/base/provider_mixins/catchup.py create mode 100644 lib/streaming_providers/base/provider_mixins/epg.py create mode 100644 lib/streaming_providers/base/provider_mixins/favorites.py create mode 100644 lib/streaming_providers/base/provider_mixins/http.py create mode 100644 lib/streaming_providers/base/provider_mixins/metadata.py create mode 100644 lib/streaming_providers/base/provider_mixins/recordings.py create mode 100644 lib/streaming_providers/base/provider_mixins/subscriptions.py create mode 100644 lib/streaming_providers/base/provider_mixins/timers.py create mode 100644 lib/streaming_providers/base/provider_mixins/vod.py diff --git a/lib/streaming_providers/base/auth_type.py b/lib/streaming_providers/base/auth_type.py new file mode 100644 index 0000000..9cb64e5 --- /dev/null +++ b/lib/streaming_providers/base/auth_type.py @@ -0,0 +1,21 @@ +"""Shared AuthType enum. + +Pulled out of provider.py into its own module so that provider.py and the +provider_mixins package can both import it without a circular import +(provider.py composes the mixins; the mixins need AuthType). + +`from streaming_providers.base.provider import AuthType` still works — +provider.py re-exports it — so no existing import site needs to change. +""" + +from enum import Enum + + +class AuthType(Enum): + """Authentication token types""" + + BEARER = "bearer" + BASIC = "basic" + CLIENT = "client" + CUSTOM = "custom" + NONE = "none" \ No newline at end of file diff --git a/lib/streaming_providers/base/provider.py b/lib/streaming_providers/base/provider.py index 639afbb..5832732 100644 --- a/lib/streaming_providers/base/provider.py +++ b/lib/streaming_providers/base/provider.py @@ -6,47 +6,55 @@ New Features: - Class attributes for static metadata (PROVIDER_LABEL, etc.) - Static methods to get metadata without instantiation - Backward compatible with existing @property methods + +Restructure note: this class used to hold every method itself (~90 methods, +~1900 lines). The capability areas — HTTP setup, auth, EPG, VOD, bookmarks, +recordings, timers, favorites, catchup, subscriptions, static metadata — have +been split into mixins under `provider_mixins/`, one file per area. +StreamingProvider composes all of them, so every existing method, property, +and signature is still available on it exactly as before; nothing that +imports `StreamingProvider` (or `AuthType`) from this module needs to change. """ import json from abc import ABC, abstractmethod from datetime import datetime -from enum import Enum -from typing import Any, Callable, ClassVar, Dict, List, Optional, Tuple +from typing import ClassVar, Dict, List, Optional, Tuple -from ..providers.auth import AuthContext, AuthStatus -from .models.proxy_models import ProxyConfig +from ..providers.auth import AuthContext, AuthStatus # noqa: F401 (re-exported for callers) +from .auth_type import AuthType # noqa: F401 (re-exported for callers) from .models import DRMConfig, Event, StreamingChannel -from .models.bookmark import Bookmark, ContentType -from .models.favorite import FavoriteType, Favorite -from .models.subscription import SubscriptionPackage, UserSubscription -from .models.recording import Recording -from .models.timer import Timer -from .models.timer_type import TimerType -from .models.epg_models import EPGEntry, EPGProgramDetails -from .network import HTTPManager, HTTPManagerFactory -from .utils.logger import logger +from .provider_mixins.auth import ProviderAuthMixin +from .provider_mixins.bookmarks import ProviderBookmarksMixin +from .provider_mixins.catchup import ProviderCatchupMixin +from .provider_mixins.epg import ProviderEpgMixin +from .provider_mixins.favorites import ProviderFavoritesMixin +from .provider_mixins.http import ProviderHttpMixin +from .provider_mixins.metadata import ProviderMetadataMixin +from .provider_mixins.recordings import ProviderRecordingsMixin +from .provider_mixins.subscriptions import ProviderSubscriptionsMixin +from .provider_mixins.timers import ProviderTimersMixin +from .provider_mixins.vod import ProviderVodMixin -class AuthType(Enum): - """Authentication token types""" - - BEARER = "bearer" - BASIC = "basic" - CLIENT = "client" - CUSTOM = "custom" - NONE = "none" - - -class StreamingProvider(ABC): +class StreamingProvider( + ABC, + ProviderMetadataMixin, + ProviderHttpMixin, + ProviderAuthMixin, + ProviderEpgMixin, + ProviderVodMixin, + ProviderBookmarksMixin, + ProviderRecordingsMixin, + ProviderTimersMixin, + ProviderFavoritesMixin, + ProviderCatchupMixin, + ProviderSubscriptionsMixin, +): """ Abstract base class for streaming providers with centralized HTTP and auth management """ - # ============================================================================ - # STATIC METADATA (NEW) - # ============================================================================ - # Class attributes for static metadata (accessible without instantiation) PROVIDER_LABEL: ClassVar[str] = "" """Base provider label without country suffix (e.g., 'Joyn', 'RTL+')""" @@ -67,136 +75,12 @@ class StreamingProvider(ABC): self._default_user_agent = "StreamingProvider/1.0" self.authenticator = None # Optional: set by concrete providers - # ============================================================================ - # STATIC METHODS FOR METADATA EXTRACTION (NEW) - # ============================================================================ - - @classmethod - def get_static_label(cls, country: str = None) -> str: - """ - Get provider label without instantiation. - - Args: - country: Optional country code for country-specific labels - - Returns: - Provider label string - """ - base_label = cls.PROVIDER_LABEL or cls.__name__.replace("Provider", "") - - if country: - # Format country code - country_upper = country.upper() - - # Special handling for common cases - if country_upper == "DE": - return f"{base_label} Germany" - elif country_upper == "AT": - return f"{base_label} Austria" - elif country_upper == "CH": - return f"{base_label} Switzerland" - else: - return f"{base_label} ({country_upper})" - - return base_label - - @classmethod - def get_static_auth_types(cls) -> List[str]: - """ - Get supported authentication types without instantiation. - - Returns: - List of supported auth type strings - """ - return cls.SUPPORTED_AUTH_TYPES.copy() - - @classmethod - def get_static_logo(cls, country: str = None) -> str: - """ - Get provider logo URL without instantiation. - - Args: - country: Optional country code for country-specific logos - - Returns: - Logo URL string - """ - return cls.PROVIDER_LOGO - - @classmethod - def get_static_supported_countries(cls) -> List[str]: - """ - Get supported countries without instantiation. - - Returns: - List of ISO country codes - """ - return cls.SUPPORTED_COUNTRIES.copy() - - @classmethod - def get_all_possible_instances(cls) -> List[Dict[str, Any]]: - """ - Get metadata for all possible instances of this provider. - - Returns: - List of instance metadata dictionaries - """ - instances = [] - - if cls.supports_multiple_countries(): - for country in cls.SUPPORTED_COUNTRIES: - instances.append( - { - "plugin": cls.__name__.lower().replace("provider", ""), - "country": country.upper(), - "label": cls.get_static_label(country), - "requires_country_suffix": True, - } - ) - else: - # Single-country provider - instances.append( - { - "plugin": cls.__name__.lower().replace("provider", ""), - "country": "DE", # Default country for single-country providers - "label": cls.get_static_label(), - "requires_country_suffix": False, - } - ) - - return instances - - # ============================================================================ - # INSTANCE PROPERTIES (Backward Compatible) - # ============================================================================ - @property @abstractmethod def provider_name(self) -> str: """Return the provider name (e.g., 'joyn', 'zdf', 'ard')""" pass - @property - def provider_label(self) -> str: - """Return the provider label (e.g., 'JOYN', 'ZDF', 'RTL+')""" - # Use static method with instance's country - return self.get_static_label(self.country) - - @property - def provider_logo(self) -> str: - """Return the provider logo URL""" - return self.get_static_logo() - - @property - def supported_auth_types(self) -> List[str]: - """List of authentication types this provider supports.""" - return self.get_static_auth_types() - - @property - def uses_dynamic_manifests(self) -> bool: - """Return True if provider uses truly dynamic manifests""" - return False - def get_channels(self, **kwargs) -> List[StreamingChannel]: """Fetch channels from the provider""" return [] @@ -231,26 +115,6 @@ class StreamingProvider(ABC): """ return [] - @property - def catchup_window(self) -> int: - """ - Return the catchup window in HOURS for this provider. - - Returns: - int: Number of hours of catchup available (0 = no catchup support) - """ - return 0 - - @property - def supports_catchup(self) -> bool: - """ - Check if provider supports catchup/timeshift functionality. - - Returns: - bool: True if catchup is supported - """ - return self.catchup_window > 0 - def enrich_channel_data( self, channel: StreamingChannel, **kwargs ) -> Optional[StreamingChannel]: @@ -304,1562 +168,4 @@ class StreamingProvider(ABC): def to_json(self, channels: List[StreamingChannel] = None, indent: int = 2) -> str: """Convert to JSON string""" - return json.dumps(self.to_output_format(channels), indent=indent, ensure_ascii=False) - - # ============================================================================ - # HTTP MANAGER SETUP (Already Implemented) - # ============================================================================ - - @property - def http_manager(self) -> Optional[HTTPManager]: - """Return the provider's HTTP manager instance""" - return self._http_manager - - @http_manager.setter - def http_manager(self, value: HTTPManager): - """Set the provider's HTTP manager instance""" - self._http_manager = value - - def _setup_http_manager( - self, - provider_name: str, - proxy_config: Optional[ProxyConfig] = None, - proxy_url: Optional[str] = None, - config_dir: Optional[str] = None, - country: Optional[str] = None, - user_agent: Optional[str] = None, - timeout: Optional[int] = None, - max_retries: Optional[int] = None, - **kwargs, - ) -> HTTPManager: - """Standard HTTP manager setup for providers with intelligent proxy resolution""" - if country is None: - country = self.country - - resolved_proxy = self._resolve_proxy_config( - proxy_config=proxy_config, - proxy_url=proxy_url, - config_dir=config_dir, - provider_name=provider_name, - country=country, - ) - - manager_kwargs = {} - if user_agent: - manager_kwargs["user_agent"] = user_agent - if timeout: - manager_kwargs["timeout"] = timeout - if max_retries: - manager_kwargs["max_retries"] = max_retries - manager_kwargs.update(kwargs) - - http_manager = HTTPManagerFactory.create_for_provider( - provider_name=provider_name, proxy_config=resolved_proxy, **manager_kwargs - ) - - self._log_http_manager_setup(provider_name, resolved_proxy, manager_kwargs) - return http_manager - - @staticmethod - def _resolve_proxy_config( - proxy_config: Optional[ProxyConfig], - proxy_url: Optional[str], - config_dir: Optional[str], - provider_name: str, - country: str, - ) -> Optional[ProxyConfig]: - """Resolve proxy configuration from multiple sources with priority""" - if proxy_config is not None: - logger.debug(f"{provider_name}: Using directly provided proxy configuration") - return proxy_config - - if proxy_url: - try: - logger.debug(f"{provider_name}: Creating proxy config from URL") - return ProxyConfig.from_url(proxy_url) - except Exception as e: - logger.warning(f"{provider_name}: Failed to parse proxy URL '{proxy_url}': {e}") - - try: - from .network import ProxyConfigManager - - proxy_mgr = ProxyConfigManager(config_dir) - managed_proxy = proxy_mgr.get_proxy_config(provider_name, country) - - if managed_proxy: - logger.debug(f"{provider_name}: Using proxy from ProxyConfigManager") - return managed_proxy - else: - logger.debug(f"{provider_name}: No proxy configuration found in ProxyConfigManager") - - except Exception as e: - logger.warning(f"{provider_name}: Could not load proxy from ProxyConfigManager: {e}") - - logger.debug(f"{provider_name}: No proxy configuration available") - return None - - @staticmethod - def _log_http_manager_setup( - provider_name: str, proxy_config: Optional[ProxyConfig], manager_kwargs: Dict - ) -> None: - """Log HTTP manager setup information""" - info_parts = [f"HTTP manager initialized for '{provider_name}'"] - - if proxy_config: - proxy_type = proxy_config.proxy_type.value if proxy_config.proxy_type else "http" - proxy_host = f"{proxy_config.host}:{proxy_config.port}" - has_auth = "authenticated" if proxy_config.auth else "no-auth" - info_parts.append(f"proxy: {proxy_type}://{proxy_host} ({has_auth})") - else: - info_parts.append("proxy: none") - - if "user_agent" in manager_kwargs: - ua_preview = ( - manager_kwargs["user_agent"][:50] + "..." - if len(manager_kwargs["user_agent"]) > 50 - else manager_kwargs["user_agent"] - ) - info_parts.append(f"user-agent: {ua_preview}") - - if "timeout" in manager_kwargs: - info_parts.append(f"timeout: {manager_kwargs['timeout']}s") - - if "max_retries" in manager_kwargs: - info_parts.append(f"retries: {manager_kwargs['max_retries']}") - - logger.info(f"{provider_name}: {', '.join(info_parts)}") - - def _share_http_manager_with_authenticator( - self, authenticator, http_manager: Optional[HTTPManager] = None - ) -> HTTPManager: - """Share HTTP manager with authenticator for consistency""" - if http_manager is None: - http_manager = self.http_manager - - if http_manager and hasattr(authenticator, "http_manager"): - if authenticator.http_manager is None: - logger.debug(f"{self.provider_name}: Sharing HTTP manager with authenticator") - authenticator.http_manager = http_manager - else: - logger.debug(f"{self.provider_name}: Using authenticator's existing HTTP manager") - http_manager = authenticator.http_manager - - return http_manager - - # ============================================================================ - # AUTHENTICATION HEADER ABSTRACTIONS - # ============================================================================ - - def _get_base_headers( - self, - user_agent: Optional[str] = None, - accept: str = "application/json", - content_type: str = "application/json", - additional_headers: Optional[Dict[str, str]] = None, - ) -> Dict[str, str]: - """ - Get base headers for API requests - - Args: - user_agent: Custom user agent (uses default if None) - accept: Accept header value - content_type: Content-Type header value - additional_headers: Additional headers to merge - - Returns: - Dictionary of HTTP headers - """ - headers = { - "User-Agent": user_agent or self._default_user_agent, - "Accept": accept, - "Content-Type": content_type, - } - - if additional_headers: - headers.update(additional_headers) - - return headers - - def _get_authenticated_headers( - self, - auth_type: AuthType = AuthType.BEARER, - token_getter: Optional[Callable[[], str]] = None, - token_key: str = "Authorization", - base_headers: Optional[Dict[str, str]] = None, - additional_headers: Optional[Dict[str, str]] = None, - **kwargs, - ) -> Dict[str, str]: - """ - Get headers with authentication token - - This is a flexible method that handles different authentication schemes - commonly used by streaming providers. - - Args: - auth_type: Type of auth (BEARER, BASIC, CLIENT, CUSTOM, NONE) - token_getter: Function to get token (uses self.authenticator.get_bearer_token if None) - token_key: Header key for token (default: 'Authorization') - base_headers: Base headers to start with (creates new if None) - additional_headers: Additional headers to add after auth - **kwargs: Arguments passed to token_getter - - Returns: - Dictionary of authenticated HTTP headers - """ - # Start with base headers or create new - headers = base_headers.copy() if base_headers else self._get_base_headers() - - # Add authentication if needed - if auth_type != AuthType.NONE: - # Get token using provided getter or default to authenticator - if token_getter: - token = token_getter() - elif self.authenticator is not None: - token = self.authenticator.get_bearer_token(**kwargs) - else: - logger.warning(f"{self.provider_name}: No token getter or authenticator available") - token = None - - # Add auth header based on type - if token: - if auth_type == AuthType.BEARER: - headers[token_key] = f"Bearer {token}" - elif auth_type == AuthType.BASIC: - headers[token_key] = f"Basic {token}" - elif auth_type == AuthType.CLIENT: - headers[token_key] = f"Client {token}" - elif auth_type == AuthType.CUSTOM: - # Custom type - just use token as-is - headers[token_key] = token - - # Add any additional headers - if additional_headers: - headers.update(additional_headers) - - return headers - - def _build_provider_headers( - self, - base_headers: Optional[Dict[str, str]] = None, - auth_type: AuthType = AuthType.NONE, - provider_headers: Optional[Dict[str, str]] = None, - **auth_kwargs, - ) -> Dict[str, str]: - """ - Build complete headers with provider-specific fields - - This is a convenience method that combines base headers, authentication, - and provider-specific headers in one call. - - Args: - base_headers: Base headers (created if None) - auth_type: Authentication type (NONE = no auth) - provider_headers: Provider-specific headers to add - **auth_kwargs: Arguments for authentication - - Returns: - Complete headers dictionary - """ - # Start with base or provided headers - headers = base_headers.copy() if base_headers else self._get_base_headers() - - # Add authentication if needed - if auth_type != AuthType.NONE: - headers = self._get_authenticated_headers( - auth_type=auth_type, base_headers=headers, **auth_kwargs - ) - - # Add provider-specific headers - if provider_headers: - headers.update(provider_headers) - - return headers - - def _add_auth_to_headers( - self, - headers: Dict[str, str], - auth_type: AuthType = AuthType.BEARER, - token_getter: Optional[Callable[[], str]] = None, - token_key: str = "Authorization", - **kwargs, - ) -> Dict[str, str]: - """ - Add authentication to existing headers (in-place modification) - - Useful when you've already built headers and just need to add auth. - - Args: - headers: Headers dictionary to modify - auth_type: Type of authentication - token_getter: Function to get token - token_key: Header key for token - **kwargs: Arguments for token_getter - - Returns: - The modified headers dictionary (same object) - """ - if auth_type == AuthType.NONE: - return headers - - # Get token - if token_getter: - token = token_getter() - elif self.authenticator is not None: - token = self.authenticator.get_bearer_token(**kwargs) - else: - logger.warning(f"{self.provider_name}: No token available for auth") - return headers - - # Add auth header - if token: - if auth_type == AuthType.BEARER: - headers[token_key] = f"Bearer {token}" - elif auth_type == AuthType.BASIC: - headers[token_key] = f"Basic {token}" - elif auth_type == AuthType.CLIENT: - headers[token_key] = f"Client {token}" - elif auth_type == AuthType.CUSTOM: - headers[token_key] = token - - return headers - - def _get_auth_token( - self, token_type: str = "bearer", force_refresh: bool = False, **kwargs - ) -> Optional[str]: - """ - Get authentication token from authenticator - - Convenience method for getting tokens with common options. - - Args: - token_type: Type of token to get ('bearer', 'device', 'persona', etc.) - force_refresh: Force token refresh - **kwargs: Additional arguments for authenticator - - Returns: - Token string or None - """ - if self.authenticator is None: - logger.warning(f"{self.provider_name}: No authenticator available") - return None - - try: - # Try to get token based on type - if token_type == "bearer": - return self.authenticator.get_bearer_token(force_refresh=force_refresh, **kwargs) - elif hasattr(self.authenticator, f"get_{token_type}_token"): - getter = getattr(self.authenticator, f"get_{token_type}_token") - return getter(force_refresh=force_refresh, **kwargs) - else: - # Default to bearer token - return self.authenticator.get_bearer_token(force_refresh=force_refresh, **kwargs) - except Exception as e: - logger.error(f"{self.provider_name}: Error getting {token_type} token: {e}") - return None - - # ============================================================================ - # EPG PROPERTIES - MUST be overridden by providers - # ============================================================================ - - @property - def epg_window(self) -> Tuple[int, int]: - """ - Return the EPG window as (past_days, future_days). - - Returns: - Tuple[int, int]: (past_days, future_days) - (0, 0) means no EPG support - - MUST be overridden by providers. - """ - return 0, 0 - - @property - def implements_epg(self) -> bool: - """Check if provider implements EPG.""" - return self.epg_window != (0, 0) - - # ============================================================================ - # EPG METHODS - Optional with sensible defaults - # ============================================================================ - - def get_epg( - self, - channel_id: str, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, - country: Optional[str] = None, - **kwargs, - ) -> List["EPGEntry"]: - """ - Get EPG data for a specific channel. - - Override if provider supports per-channel EPG. - Default returns empty list (no EPG). - """ - return [] - - def get_epg_grid( - self, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, - channel_ids: Optional[List[str]] = None, - country: Optional[str] = None, - **kwargs, - ) -> Dict[str, List["EPGEntry"]]: - """ - Get EPG data for multiple channels in one operation. - - Override if provider supports batch EPG. - Default returns empty dict (no batch EPG). - - Note: If provider only supports per-channel EPG, - implement get_epg() and leave this as default. - """ - return {} - - def get_program_details(self, program_id: str, **kwargs) -> Optional["EPGProgramDetails"]: - """ - Get detailed metadata for a single program. - - Override if provider supports program details. - Default returns None (no details). - """ - return None - - def get_epg_xmltv(self, country: Optional[str] = None, **kwargs) -> Optional[str]: - """ - Get complete EPG data in XMLTV format. - - Override if provider supports XMLTV export. - Default returns None (no XMLTV). - """ - return None - - # ========================================================================= - # VOD - # ========================================================================= - - @property - def implements_vod(self) -> bool: - """ - Indicates whether this provider has a browsable VOD catalogue. - - Return False (and let get_vod_category return []) for providers that - only offer live channels or events. VodOperations will skip providers - where this returns False when aggregating across all providers. - """ - return False - - def get_vod_category(self, content_id: str = "", **kwargs) -> List: - """ - Return the children of a VOD tree node. - - Args: - content_id: Opaque node identifier returned by a previous - get_vod_category call. Empty string → root level. - Providers define their own ID format; the caller - treats it as an opaque token and never parses it. - - Returns: - Mixed list of VodCategory and VodItem objects. - Return [] if the node has no children or VOD is not supported. - """ - return [] - - def search_vod( - self, - query: str, - cursor: Optional[str] = None, - page_size: int = 24, - **kwargs, - ) -> List: - """ - Search the VOD catalogue for items matching query. - - Args: - query: Free-text search string entered by the user. - cursor: Opaque continuation token from a previous response's - next_cursor field. None → first page. - page_size: Hint for how many entries to return per page. - Providers may ignore or clamp this value. - - Returns: - Mixed list of VodCategory and VodItem objects, or a paged dict - with the same shape as get_vod_category (entries, next_cursor, - total). Return [] if search is not supported. - """ - return [] - - # ========================================================================= - # BOOKMARKS - # ========================================================================= - - @property - def implements_bookmarks(self) -> bool: - """ - True if this provider can store/retrieve playback positions. - - Return False (default) for providers that do not have bookmark / - continue-watching functionality. BookmarkOperations will skip - providers where this returns False when aggregating across all - providers. - - Providers that support bookmarks should override this to return True - and implement get_bookmarks(), update_bookmark(), and - delete_bookmark(). - """ - return False - - def get_bookmarks(self, **kwargs) -> List[Bookmark]: - """ - Return all bookmarks for the authenticated user. - - This method should fetch the user's continue-watching list or - playback positions from the provider's backend. - - Args: - **kwargs: Provider-specific filtering options (e.g., content_type, - limit, offset). - - Returns: - List of Bookmark objects, or [] if bookmarks are not supported - or none exist. - - Note: - The returned Bookmark objects should have their content_type field - properly set (LIVE, VOD, EVENT, RECORDING, etc.) so that the - client can correctly resolve the content. - - Default implementation returns an empty list. Override in provider - plugins that support bookmark storage. - """ - return [] - - def update_bookmark( - self, - content_id: str, - position_seconds: int, - content_type: ContentType, - duration_seconds: Optional[int] = None, - title: Optional[str] = None, - **kwargs, - ) -> Bookmark: - """ - Save or update a bookmark for specific content. - - This method is called automatically when playback stops or pauses. - The provider should store the position and associate it with the - authenticated user. - - Args: - content_id: The content being watched (channel ID, VOD ID, - etc.). - position_seconds: Where playback stopped in seconds from start. - 0 = not started / start of content. - -1 = explicitly marked as completed. - Content is also considered complete once position - reaches the model's COMPLETION_THRESHOLD - (>=95% by default). - content_type: Type of content being bookmarked. Required — - callers always know what they are bookmarking, - so this is never None. - duration_seconds: Total duration of the content (optional but - recommended for progress calculations). - title: Content title for caching (optional — provider - may ignore and use its own metadata store). - **kwargs: Provider-specific arguments (e.g., episode - number, season number, series ID). - - Returns: - The saved Bookmark object as confirmed by the provider. - - Raises: - RuntimeError: If the provider rejects the bookmark (e.g. user not - authenticated, content not accessible). - - Note: - At minimum the provider should persist content_id and - position_seconds so that get_bookmarks() can later return this - bookmark. - - Default implementation raises NotImplementedError so misconfigured - providers fail loudly rather than silently doing nothing. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement update_bookmark(). " - "Override this method to support bookmark storage." - ) - - def delete_bookmark(self, content_id: str, **kwargs) -> None: - """ - Delete a bookmark from the provider's backend. - - Called when: - - User manually removes a bookmark from "Continue Watching" - - Content is removed from the provider - - Cleanup of stale bookmarks - - Args: - content_id: The content identifier whose bookmark should be - removed. - **kwargs: Provider-specific arguments. - - Returns: - None on success. - - Raises: - KeyError: If no bookmark with this content_id exists. - RuntimeError: If the provider refuses deletion (e.g. permission - denied, backend error). - - Note: - Deleting a non-existent bookmark must raise KeyError rather than - silently succeeding, so that callers can distinguish "already - gone" from "successfully deleted". - - Default implementation raises NotImplementedError so misconfigured - providers fail loudly. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement delete_bookmark(). " - "Override this method to support bookmark deletion." - ) - - # ========================================================================= - # BOOKMARK HELPER METHODS (Optional overrides) - # ========================================================================= - - def batch_update_bookmarks( - self, updates: List[Dict[str, Any]], **kwargs - ) -> List[Bookmark]: - """ - Update multiple bookmarks in a single batch operation. - - Useful for synchronising local bookmark state with the provider - backend, or for bulk writes after a playback session. - - Args: - updates: List of dictionaries, each containing:: - - { - "content_id": str, - "position_seconds": int, - "content_type": ContentType, - "duration_seconds": Optional[int], - "title": Optional[str], - ... (other provider-specific kwargs) - } - - **kwargs: Provider-specific batch options passed to every - individual update_bookmark() call. - - Returns: - List of updated Bookmark objects in the same order as ``updates``. - If an individual update fails, the exception propagates and the - list contains only the bookmarks that succeeded before the - failure. - - Raises: - RuntimeError: If the batch operation fails partially or completely. - - Note: - The default implementation falls back to individual - update_bookmark() calls and does **not** mutate the dicts in - ``updates``. Override if your provider supports a native batch - endpoint for efficiency. - """ - results = [] - for raw in updates: - # Work on a copy so the caller's dicts are never mutated. - update = raw.copy() - content_id = update.pop("content_id") - position_seconds = update.pop("position_seconds") - content_type = update.pop("content_type") - duration_seconds = update.pop("duration_seconds", None) - title = update.pop("title", None) - - result = self.update_bookmark( - content_id=content_id, - position_seconds=position_seconds, - content_type=content_type, - duration_seconds=duration_seconds, - title=title, - **update, - **kwargs, - ) - results.append(result) - - return results - - # ========================================================================= - # RECORDINGS - # ========================================================================= - - @property - def implements_recordings(self) -> bool: - """ - True if this provider can return recorded content. - - Return False (default) for providers that only offer live or VOD content. - RecordingOperations skips providers where this returns False when - aggregating across all providers. - """ - return False - - def get_recordings( - self, - include_deleted: bool = False, - **kwargs, - ) -> List[Recording]: - """ - Return recordings available for the authenticated user. - - Args: - include_deleted: If True, also return recordings marked as deleted - (useful for a trash/recycle-bin view). - **kwargs: Provider-specific filtering options. - - Returns: - List of Recording objects, or [] if recordings are not supported. - """ - return [] - - def delete_recording(self, recording_id: str, **kwargs) -> None: - """ - Permanently delete a recording on the provider's backend. - - Args: - recording_id: The recording to delete (== content_id on the model). - **kwargs: Provider-specific arguments. - - Returns: - None on success. - - Raises: - KeyError: If no recording with this ID exists. - RuntimeError: If the provider refuses the deletion (e.g. the - recording is currently being captured, or the user - lacks permission). - - Default implementation raises NotImplementedError so misconfigured - providers fail loudly rather than silently doing nothing. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement delete_recording(). " - "Override this method to support recording deletion." - ) - - # ========================================================================= - # TIMERS - # ========================================================================= - - @property - def implements_timers(self) -> bool: - """ - True if this provider supports scheduled recording timers. - - Return False (default) for providers that only offer live or VOD content - and have no PVR/timer backend. TimerOperations skips providers where - this returns False when aggregating across all providers. - """ - return False - - def get_timer_types(self) -> List["TimerType"]: - """ - Return the timer types this provider supports. - - Providers that support timers MUST override this method and return at - least one TimerType so that clients know what fields to present when - creating a timer. - - Returns: - List of TimerType objects, or [] if timers are not supported. - """ - return [] - - def get_timers(self, **kwargs) -> List["Timer"]: - """ - Return all timers (scheduled recordings) for the authenticated user. - - Args: - **kwargs: Provider-specific filtering options. - - Returns: - List of Timer objects, or [] if timers are not supported. - """ - return [] - - def add_timer(self, timer: "Timer", **kwargs) -> "Timer": - """ - Schedule a new timer on the provider's backend. - - Args: - timer: Timer to create. timer.client_index is ignored — the - provider allocates and sets it on the returned object. - **kwargs: Provider-specific arguments. - - Returns: - The saved Timer with client_index populated by the provider. - - Raises: - RuntimeError: If the provider rejects the timer (e.g. scheduling - conflict, unsupported timer type, insufficient - permissions). - - Default raises NotImplementedError so misconfigured providers fail - loudly rather than silently doing nothing. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement add_timer(). " - "Override this method to support timer creation." - ) - - def update_timer(self, timer: "Timer", **kwargs) -> "Timer": - """ - Update an existing timer on the provider's backend. - - Args: - timer: Timer with updated fields. timer.client_index identifies - the record to modify. - **kwargs: Provider-specific arguments. - - Returns: - The updated Timer as confirmed by the provider. - - Raises: - KeyError: If no timer with that client_index exists. - RuntimeError: If the provider refuses the update (e.g. the timer - is currently recording). - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement update_timer(). " - "Override this method to support timer updates." - ) - - def delete_timer( - self, client_index: int, force_delete: bool = False, **kwargs - ) -> None: - """ - Delete a timer on the provider's backend. - - Args: - client_index: Timer identifier to delete. - force_delete: If True and the timer is currently recording, abort - the ongoing capture before deleting. - **kwargs: Provider-specific arguments. - - Returns: - None on success. - - Raises: - KeyError: If no timer with that client_index exists. - RuntimeError: If the provider refuses deletion (e.g. recording in - progress and force_delete is False). - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement delete_timer(). " - "Override this method to support timer deletion." - ) - - # ============================================================================ - # Favorite Management - # ============================================================================ - - @property - def implements_favorites(self) -> bool: - """ - True if this provider can store/retrieve user favorites. - - Default False. Override in providers that support favorites. - """ - return False - - def get_favorites(self, **kwargs) -> List["Favorite"]: - """ - Return all favorites for the authenticated user. - - Returns: - List of Favorite objects, or [] if not supported or none exist. - """ - return [] - - def add_favorite( - self, - content_id: str, - favorite_type: FavoriteType, - title: Optional[str] = None, - **kwargs, - ) -> "Favorite": - """ - Add a content to user's favorites. - - Args: - content_id: Content identifier. - favorite_type: Type of content (PROGRAM, CLIP, LIVE, EVENT). - title: Content title (optional for caching). - - Returns: - The saved Favorite object. - - Raises: - RuntimeError: If the provider rejects the favorite. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement add_favorite()." - ) - - def remove_favorite(self, content_id: str, **kwargs) -> None: - """ - Remove a content from user's favorites. - - Args: - content_id: Content identifier to remove. - - Raises: - KeyError: If no favorite with this content_id exists. - RuntimeError: If the provider refuses deletion. - """ - raise NotImplementedError( - f"{self.__class__.__name__} does not implement remove_favorite()." - ) - - # ============================================================================ - # CATCHUP ABSTRACT METHODS - # ============================================================================ - - def get_catchup_manifest( - self, - content_id: str, - start_time: int, - end_time: int, - epg_id: Optional[str] = None, - **kwargs, - ) -> Optional[str]: - """ - Get manifest URL for catchup/timeshift content. - - Args: - content_id: Channel identifier - start_time: Start time as Unix timestamp - end_time: End time as Unix timestamp - epg_id: Optional EPG event ID (might be needed by some providers) - **kwargs: Additional provider-specific parameters - - Returns: - Manifest URL for catchup content, or None if not supported - - Default implementation raises NotImplementedError. - Override in subclass to implement provider-specific catchup logic. - - Do NOT fall back to self.get_manifest() here — returning the live - manifest URL as a catchup manifest will cause the DRM pipeline to - extract PSSH from the live stream, which may differ from the catchup - stream's encryption context. - """ - raise NotImplementedError( - f"{self.__class__.__name__}.get_catchup_manifest() is not implemented." - ) - - def get_catchup_manifest_headers( - self, - content_id: str, - start_time: int, - end_time: int, - epg_id: Optional[str] = None, - **kwargs, - ) -> Dict[str, str]: - """ - Return HTTP headers for the catchup manifest request. - - Default implementation delegates to get_manifest_headers() since many - providers use the same auth headers for live and catchup manifests. - Override when catchup requires different headers (e.g. extra tokens). - """ - return self.get_manifest_headers(content_id, **kwargs) - - def get_catchup_manifest_with_headers( - self, - content_id: str, - start_time: int, - end_time: int, - epg_id: Optional[str] = None, - **kwargs, - ) -> Tuple[Optional[str], Dict[str, str]]: - """ - Convenience method returning (catchup_manifest_url, headers). - - This is the single entry point used by CatchupOperations before - calling into the DRM pipeline, mirroring the role that - get_manifest_with_headers() plays for live content. - - Providers should override get_catchup_manifest() (and optionally - get_catchup_manifest_headers()) rather than this method directly. - - Raises: - NotImplementedError: propagated from get_catchup_manifest() if the - provider has not implemented catchup manifest resolution. - """ - url = self.get_catchup_manifest( - content_id=content_id, - start_time=start_time, - end_time=end_time, - epg_id=epg_id, - **kwargs, - ) - headers = self.get_catchup_manifest_headers( - content_id=content_id, - start_time=start_time, - end_time=end_time, - epg_id=epg_id, - **kwargs, - ) - return url, headers - - def get_catchup_drm( - self, - content_id: str, - start_time: int, - end_time: int, - epg_id: Optional[str] = None, - drm_variant: Optional[str] = None, - **kwargs, - ) -> List[DRMConfig]: - """ - Get DRM configurations for catchup content. - - Args: - content_id: Channel identifier - start_time: Start time as Unix timestamp - end_time: End time as Unix timestamp - epg_id: Optional EPG event ID (might be needed for DRM licensing) - drm_variant: Optional DRM variant ('auto', 'software', 'hardware') - **kwargs: Additional provider-specific parameters - - Returns: - List of DRM configurations for catchup content - - Default implementation raises NotImplementedError so that the DRM - pipeline falls through to PSSH extraction from the catchup manifest. - - Override in subclass when catchup requires a *different* DRM - configuration from live (e.g. a different license URL, extra request - headers, or a static ClearKey set). If catchup uses exactly the same - DRM as live, implement as: - - def get_catchup_drm(self, content_id, start_time, end_time, - epg_id=None, drm_variant=None, **kwargs): - return self.get_drm(content_id, drm_variant=drm_variant, **kwargs) - - Do NOT call super().get_drm() silently — that would make the pipeline - think Phase 2 produced valid configs from the live stream context, - which is wrong when the catchup manifest has different encryption. - """ - raise NotImplementedError( - f"{self.__class__.__name__}.get_catchup_drm() is not implemented. " - "The DRM pipeline will extract PSSH from the catchup manifest directly. " - "Override this method only if catchup requires a custom DRM configuration." - ) - - # ============================================================================ - # CATCHUP HELPER METHODS - # ============================================================================ - - def get_catchup_window_for_channel(self, content_id: str) -> int: - """ - Get catchup window for a specific channel in HOURS. - - Args: - content_id: Channel identifier - - Returns: - int: Catchup window in hours for this channel - """ - return self.catchup_window - - def validate_catchup_request( - self, start_time: int, end_time: int - ) -> tuple[bool, Optional[str]]: - """ - Validate a catchup request against provider's capabilities. - - Args: - start_time: Start time as Unix timestamp - end_time: End time as Unix timestamp - - Returns: - Tuple of (is_valid, error_message) - """ - import time - - if not self.supports_catchup: - return False, f"Provider '{self.provider_name}' does not support catchup" - - if start_time >= end_time: - return False, "Invalid time range: start_time must be before end_time" - - now = int(time.time()) - if start_time > now: - return False, "Cannot request future content" - - # CHANGE FROM DAYS TO HOURS HERE - max_age_seconds = self.catchup_window * 3600 # hours to seconds - content_age = now - start_time - - if content_age > max_age_seconds: - hours_ago = content_age // 3600 - return False, ( - f"Content is outside catchup window " - f"(requested: {hours_ago} hours ago, " - f"max: {self.catchup_window} hours)" - ) - - return True, None - - def format_catchup_time_params( - self, start_time: int, end_time: int, format_type: str = "iso" - ) -> Dict[str, str]: - """ - Format time parameters for provider-specific API calls. - - Different providers expect different time formats in their APIs. - This helper converts Unix timestamps to various formats. - - Args: - start_time: Start time as Unix timestamp - end_time: End time as Unix timestamp - format_type: Format type ('iso', 'unix', 'millis', 'custom') - - Returns: - Dictionary with formatted time parameters - - Override in subclass for provider-specific formatting. - """ - from datetime import datetime - - if format_type == "iso": - # ISO 8601 format - start_dt = datetime.fromtimestamp(start_time) - end_dt = datetime.fromtimestamp(end_time) - return {"start": start_dt.isoformat(), "end": end_dt.isoformat()} - elif format_type == "unix": - # Unix timestamps (seconds) - return {"start": str(start_time), "end": str(end_time)} - elif format_type == "millis": - # Milliseconds since epoch - return {"start": str(start_time * 1000), "end": str(end_time * 1000)} - else: - # Default to unix - return {"start": str(start_time), "end": str(end_time)} - - def build_catchup_manifest_url( - self, base_url: str, start_time: int, end_time: int, url_format: str = "query" - ) -> str: - """ - Build catchup manifest URL with time parameters. - - Helper method to construct manifest URLs with time parameters - in various formats that different providers use. - - Args: - base_url: Base manifest URL - start_time: Start time as Unix timestamp - end_time: End time as Unix timestamp - url_format: Format ('query', 'path', 'fragment') - - Returns: - Complete manifest URL with time parameters - - Override in subclass for provider-specific URL construction. - """ - if url_format == "query": - # Add as query parameters - separator = "&" if "?" in base_url else "?" - return f"{base_url}{separator}start={start_time}&end={end_time}" - elif url_format == "path": - # Add to path (e.g., /manifest/start/end.mpd) - return f"{base_url}/{start_time}/{end_time}" - elif url_format == "fragment": - # Add as URL fragment (e.g., manifest.mpd#t=start,end) - return f"{base_url}#t={start_time},{end_time}" - else: - # Default to query parameters - separator = "&" if "?" in base_url else "?" - return f"{base_url}{separator}start={start_time}&end={end_time}" - - # ============================================================================ - # SUBSCRIPTION METHODS - # ============================================================================ - - def get_subscription_status(self, **kwargs) -> Optional[UserSubscription]: - """ - Get user's subscription status for this provider. - - Returns: - UserSubscription object with subscription details, or - None if provider doesn't support subscription queries or - subscription info isn't available. - - Default implementation returns None (subscription not supported). - Override in provider plugins that support subscription checking. - - Example usage in providers: - # Query provider API for user entitlements - # Parse response into SubscriptionPackage objects - # Return UserSubscription with packages and accessible channels - """ - return None - - def get_subscribed_channels(self, **kwargs) -> List[StreamingChannel]: - """ - Get channels the current user is subscribed to. - - This method: - 1. Gets subscription status (if supported) - 2. Filters all channels based on accessible channel IDs - 3. Returns filtered list or all channels as fallback - - Returns: - List of StreamingChannel objects that the user can access - - Note: - Override get_subscription_status() in provider plugins - to enable subscription filtering. - """ - # Get all channels first - all_channels = self.get_channels(**kwargs) - - # Try to get subscription status - subscription = self.get_subscription_status(**kwargs) - - # If no subscription info or not active, return all channels - if not subscription or not subscription.active: - return all_channels - - # Filter channels based on accessible channel IDs - if subscription.accessible_channel_ids: - return [ - channel - for channel in all_channels - if channel.channel_id in subscription.accessible_channel_ids - ] - - # No filtering possible, return all channels - return all_channels - - def get_available_packages(self, **kwargs) -> List[SubscriptionPackage]: - """ - Get all subscription packages available from this provider. - - Useful for: - - Displaying upgrade options in UI - - Showing package comparison - - Subscription management interface - - Returns: - List of available SubscriptionPackage objects - Empty list if not implemented or no packages available - - Default implementation returns empty list. - Override in provider plugins that have package information. - """ - return [] - - def is_channel_accessible(self, channel_id: str, **kwargs) -> bool: - """ - Check if a specific channel is accessible with current subscription. - - Args: - channel_id: ID of the channel to check - **kwargs: Additional arguments passed to get_subscription_status() - - Returns: - True if channel is accessible, False otherwise. - Returns True if subscription checking is not supported. - - Note: - This is a convenience method for quick checks. - """ - subscription = self.get_subscription_status(**kwargs) - - # If no subscription info, assume accessible (backward compatibility) - if not subscription or not subscription.active: - return True - - # Check if channel is in accessible set - return subscription.can_access_channel(channel_id) - - @classmethod - def get_supported_countries(cls) -> List[str]: - """ - Get list of countries supported by this provider. - - Returns: - List of ISO country codes (e.g., ['de', 'at', 'ch']) - Empty list means single-country provider using default country - """ - return cls.SUPPORTED_COUNTRIES.copy() - - @classmethod - def supports_multiple_countries(cls) -> bool: - """ - Check if this provider supports multiple countries. - - Returns: - True if provider supports country-specific instances - """ - return len(cls.SUPPORTED_COUNTRIES) > 1 - - @classmethod - def validate_country(cls, country: str) -> bool: - """ - Validate if a country is supported by this provider. - - Args: - country: ISO country code to validate - - Returns: - True if country is supported or provider is single-country - """ - if not cls.supports_multiple_countries(): - # Single-country providers accept any country (or ignore it) - return True - - return country.lower() in [c.lower() for c in cls.SUPPORTED_COUNTRIES] - - def validate_auth_type(self, auth_type: str) -> bool: - """ - Check if an auth type is supported by this provider. - - Useful for: - - Validating user input in configuration UI - - Safely switching auth modes - - Error messages when unsupported auth is requested - - Args: - auth_type: Auth type to check (e.g., 'user_credentials') - - Returns: - True if supported, False otherwise - - Example: - if provider.validate_auth_type('user_credentials'): - # Safe to request user credentials - """ - return auth_type in self.supported_auth_types - - def get_auth_type_description(self, auth_type: str) -> str: - """ - Get human-readable description of an auth type. - - Args: - auth_type: Auth type to describe - - Returns: - Description string or empty string if not supported - """ - descriptions = { - "user_credentials": "Username and password authentication", - "client_credentials": "Client ID and secret authentication", - "network_based": "Network/fixed-line authentication", - "anonymous": "No authentication required", - "device_registration": "Device registration authentication", - "embedded_client": "Built-in credentials authentication", - } - - if auth_type in descriptions: - return descriptions[auth_type] - - # For custom auth types - return f"Custom authentication: {auth_type}" - - def get_auth_requirements(self, auth_type: str) -> Dict[str, Any]: - """ - Get requirements for a specific auth type. - - Args: - auth_type: Auth type to get requirements for - - Returns: - Dictionary with requirement information - - Raises: - ValueError: If auth_type is not supported - """ - if not self.validate_auth_type(auth_type): - raise ValueError(f"Auth type '{auth_type}' not supported by {self.provider_name}") - - requirements = { - "auth_type": auth_type, - "needs_storage": auth_type in ["user_credentials", "client_credentials"], - "provides_token": auth_type != "anonymous", - "user_interaction_required": auth_type in ["user_credentials", "device_registration"], - } - - # Type-specific details - if auth_type == "user_credentials": - requirements.update( - { - "fields": ["username", "password"], - "optional_fields": ["client_id"], - "storage_key": "user_password", - } - ) - elif auth_type == "client_credentials": - requirements.update( - { - "fields": ["client_id", "client_secret"], - "storage_key": "client_credentials", - } - ) - elif auth_type == "network_based": - requirements.update( - { - "description": "Authenticates via your network provider", - "automatic": True, - } - ) - - return requirements - - # ===== AUTHENTICATION PROPERTIES AND METHODS ===== - - @property - def preferred_auth_type(self) -> str: - """Preferred authentication type (first in supported list).""" - types = self.supported_auth_types - return types[0] if types else "unknown" - - @property - def requires_stored_credentials(self) -> bool: - """True if provider needs credentials stored in settings.""" - credential_types = ["user_credentials", "client_credentials"] - return any(auth_type in credential_types for auth_type in self.supported_auth_types) - - @property - def requires_manifest_context(self) -> bool: - """True if provider needs to use same http manager to get manifest.""" - return False - - # ===== AUTHENTICATION PROPERTIES ===== - - def get_current_auth_type(self, context: AuthContext) -> str: - """ - Determine which auth type is currently active. - - Default implementation checks tokens/credentials. - Override for providers with complex auth logic. - - Args: - context: AuthContext for accessing tokens/credentials - - Returns: - Current active auth type - """ - return self._determine_current_auth_type_default(context) - - def _determine_current_auth_type_default(self, context: AuthContext) -> str: - """ - Default logic for determining current auth type. - Providers can override get_current_auth_type() directly instead. - """ - # 1. Check if provider requires stored credentials - if self.requires_stored_credentials: - credentials = context.get_credentials(self.provider_name, self.country) - if credentials: - # Map credential type to auth type - if hasattr(credentials, "credential_type"): - if credentials.credential_type == "user_password": - return "user_credentials" - elif credentials.credential_type == "client_credentials": - return "client_credentials" - - # 2. Check token auth level - primary_token = context.get_token( - self.provider_name, self.primary_token_scope, self.country - ) - if primary_token: - auth_level = primary_token.get("auth_level") - if auth_level == "user_authenticated": - return "user_credentials" - elif auth_level == "client_credentials": - return "client_credentials" - elif auth_level == "anonymous": - return "anonymous" - elif auth_level == "network_based": - return "network_based" - - # 3. Return first supported type as default - return self.preferred_auth_type - - # Token management properties (keep these) - @property - def primary_token_scope(self) -> Optional[str]: - """ - Primary token scope for this provider. - None = uses root-level token or no token needed. - - Returns: - Token scope string or None - """ - return None - - @property - def token_scopes(self) -> List[str]: - """ - All token scopes this provider uses. - - Returns: - List of token scope strings - """ - scope = self.primary_token_scope - return [scope] if scope else [] - - def get_auth_status(self, context: AuthContext) -> "AuthStatus": - """ - Get authentication status for this provider. - Uses AuthStatusBuilder by default. - - Override only for providers with special requirements. - - Args: - context: AuthContext with access to settings - - Returns: - AuthStatus object - """ - from ..providers.auth_builder import ( - AuthStatusBuilder, - ) # Import here to avoid circular imports - - return AuthStatusBuilder.for_provider(self, context) - - # Optional override methods for providers with special logic - def _calculate_auth_state(self, context: AuthContext): - """ - Override to provide custom auth state calculation. - Return None to use standard calculation. - - Returns: - AuthState or None - """ - return None - - def _calculate_readiness(self, context: AuthContext): - """ - Override to provide custom readiness calculation. - Return None to use standard calculation. - - Returns: - Tuple of (is_ready: bool, reason: str) or None - """ - return None - - def get_auth_details(self, context: AuthContext) -> Dict[str, Any]: - """ - Override to provide provider-specific auth details. - - Returns: - Dictionary with provider-specific information - """ - return {} \ No newline at end of file + return json.dumps(self.to_output_format(channels), indent=indent, ensure_ascii=False) \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/auth.py b/lib/streaming_providers/base/provider_mixins/auth.py new file mode 100644 index 0000000..6dcf310 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/auth.py @@ -0,0 +1,483 @@ +"""Provider authentication mixin: header building, token retrieval, auth-type +introspection and AuthStatus reporting. + +Behavior-preserving split out of provider.py, with one fix: the BEARER/BASIC/ +CLIENT/CUSTOM header-formatting switch used to be duplicated verbatim in both +`_get_authenticated_headers` and `_add_auth_to_headers`. It's now a single +`_format_auth_header_value` helper that both call — same output for the same +input, one place to extend when a new AuthType is added. +""" + +from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional + +from ...providers.auth import AuthContext, AuthStatus +from ..auth_type import AuthType +from ..utils.logger import logger + + +class ProviderAuthMixin: + if TYPE_CHECKING: + # Provided by StreamingProvider / sibling mixins once this is + # actually composed into the full class. Declared here only so + # IDEs and type checkers stop flagging them as unresolved on + # ProviderAuthMixin in isolation — no runtime effect. + provider_name: str + country: str + authenticator: Any + _default_user_agent: str + supported_auth_types: List[str] + + def _get_base_headers( + self, + user_agent: Optional[str] = None, + accept: str = "application/json", + content_type: str = "application/json", + additional_headers: Optional[Dict[str, str]] = None, + ) -> Dict[str, str]: + """ + Get base headers for API requests + + Args: + user_agent: Custom user agent (uses default if None) + accept: Accept header value + content_type: Content-Type header value + additional_headers: Additional headers to merge + + Returns: + Dictionary of HTTP headers + """ + headers = { + "User-Agent": user_agent or self._default_user_agent, + "Accept": accept, + "Content-Type": content_type, + } + + if additional_headers: + headers.update(additional_headers) + + return headers + + @staticmethod + def _format_auth_header_value(auth_type: AuthType, token: str) -> Optional[str]: + """ + Render a token into the header value for the given auth type. + + Shared by `_get_authenticated_headers` and `_add_auth_to_headers` so + the BEARER/BASIC/CLIENT/CUSTOM mapping only lives in one place. + + Returns None for an auth type that isn't one of the known schemes + (callers leave the header untouched in that case, matching the + previous behavior of simply not entering any elif branch). + """ + if auth_type == AuthType.BEARER: + return f"Bearer {token}" + elif auth_type == AuthType.BASIC: + return f"Basic {token}" + elif auth_type == AuthType.CLIENT: + return f"Client {token}" + elif auth_type == AuthType.CUSTOM: + return token + return None + + def _get_authenticated_headers( + self, + auth_type: AuthType = AuthType.BEARER, + token_getter: Optional[Callable[[], str]] = None, + token_key: str = "Authorization", + base_headers: Optional[Dict[str, str]] = None, + additional_headers: Optional[Dict[str, str]] = None, + **kwargs, + ) -> Dict[str, str]: + """ + Get headers with authentication token + + This is a flexible method that handles different authentication schemes + commonly used by streaming providers. + + Args: + auth_type: Type of auth (BEARER, BASIC, CLIENT, CUSTOM, NONE) + token_getter: Function to get token (uses self.authenticator.get_bearer_token if None) + token_key: Header key for token (default: 'Authorization') + base_headers: Base headers to start with (creates new if None) + additional_headers: Additional headers to add after auth + **kwargs: Arguments passed to token_getter + + Returns: + Dictionary of authenticated HTTP headers + """ + # Start with base headers or create new + headers = base_headers.copy() if base_headers else self._get_base_headers() + + # Add authentication if needed + if auth_type != AuthType.NONE: + # Get token using provided getter or default to authenticator + if token_getter: + token = token_getter() + elif self.authenticator is not None: + token = self.authenticator.get_bearer_token(**kwargs) + else: + logger.warning(f"{self.provider_name}: No token getter or authenticator available") + token = None + + # Add auth header based on type + if token: + value = self._format_auth_header_value(auth_type, token) + if value is not None: + headers[token_key] = value + + # Add any additional headers + if additional_headers: + headers.update(additional_headers) + + return headers + + def _build_provider_headers( + self, + base_headers: Optional[Dict[str, str]] = None, + auth_type: AuthType = AuthType.NONE, + provider_headers: Optional[Dict[str, str]] = None, + **auth_kwargs, + ) -> Dict[str, str]: + """ + Build complete headers with provider-specific fields + + This is a convenience method that combines base headers, authentication, + and provider-specific headers in one call. + + Args: + base_headers: Base headers (created if None) + auth_type: Authentication type (NONE = no auth) + provider_headers: Provider-specific headers to add + **auth_kwargs: Arguments for authentication + + Returns: + Complete headers dictionary + """ + # Start with base or provided headers + headers = base_headers.copy() if base_headers else self._get_base_headers() + + # Add authentication if needed + if auth_type != AuthType.NONE: + headers = self._get_authenticated_headers( + auth_type=auth_type, base_headers=headers, **auth_kwargs + ) + + # Add provider-specific headers + if provider_headers: + headers.update(provider_headers) + + return headers + + def _add_auth_to_headers( + self, + headers: Dict[str, str], + auth_type: AuthType = AuthType.BEARER, + token_getter: Optional[Callable[[], str]] = None, + token_key: str = "Authorization", + **kwargs, + ) -> Dict[str, str]: + """ + Add authentication to existing headers (in-place modification) + + Useful when you've already built headers and just need to add auth. + + Args: + headers: Headers dictionary to modify + auth_type: Type of authentication + token_getter: Function to get token + token_key: Header key for token + **kwargs: Arguments for token_getter + + Returns: + The modified headers dictionary (same object) + """ + if auth_type == AuthType.NONE: + return headers + + # Get token + if token_getter: + token = token_getter() + elif self.authenticator is not None: + token = self.authenticator.get_bearer_token(**kwargs) + else: + logger.warning(f"{self.provider_name}: No token available for auth") + return headers + + # Add auth header + if token: + value = self._format_auth_header_value(auth_type, token) + if value is not None: + headers[token_key] = value + + return headers + + def _get_auth_token( + self, token_type: str = "bearer", force_refresh: bool = False, **kwargs + ) -> Optional[str]: + """ + Get authentication token from authenticator + + Convenience method for getting tokens with common options. + + Args: + token_type: Type of token to get ('bearer', 'device', 'persona', etc.) + force_refresh: Force token refresh + **kwargs: Additional arguments for authenticator + + Returns: + Token string or None + """ + if self.authenticator is None: + logger.warning(f"{self.provider_name}: No authenticator available") + return None + + try: + # Try to get token based on type + if token_type == "bearer": + return self.authenticator.get_bearer_token(force_refresh=force_refresh, **kwargs) + elif hasattr(self.authenticator, f"get_{token_type}_token"): + getter = getattr(self.authenticator, f"get_{token_type}_token") + return getter(force_refresh=force_refresh, **kwargs) + else: + # Default to bearer token + return self.authenticator.get_bearer_token(force_refresh=force_refresh, **kwargs) + except Exception as e: + logger.error(f"{self.provider_name}: Error getting {token_type} token: {e}") + return None + + def validate_auth_type(self, auth_type: str) -> bool: + """ + Check if an auth type is supported by this provider. + + Useful for: + - Validating user input in configuration UI + - Safely switching auth modes + - Error messages when unsupported auth is requested + + Args: + auth_type: Auth type to check (e.g., 'user_credentials') + + Returns: + True if supported, False otherwise + + Example: + if provider.validate_auth_type('user_credentials'): + # Safe to request user credentials + """ + return auth_type in self.supported_auth_types + + def get_auth_type_description(self, auth_type: str) -> str: + """ + Get human-readable description of an auth type. + + Args: + auth_type: Auth type to describe + + Returns: + Description string or empty string if not supported + """ + descriptions = { + "user_credentials": "Username and password authentication", + "client_credentials": "Client ID and secret authentication", + "network_based": "Network/fixed-line authentication", + "anonymous": "No authentication required", + "device_registration": "Device registration authentication", + "embedded_client": "Built-in credentials authentication", + } + + if auth_type in descriptions: + return descriptions[auth_type] + + # For custom auth types + return f"Custom authentication: {auth_type}" + + def get_auth_requirements(self, auth_type: str) -> Dict[str, Any]: + """ + Get requirements for a specific auth type. + + Args: + auth_type: Auth type to get requirements for + + Returns: + Dictionary with requirement information + + Raises: + ValueError: If auth_type is not supported + """ + if not self.validate_auth_type(auth_type): + raise ValueError(f"Auth type '{auth_type}' not supported by {self.provider_name}") + + requirements: Dict[str, Any] = { + "auth_type": auth_type, + "needs_storage": auth_type in ["user_credentials", "client_credentials"], + "provides_token": auth_type != "anonymous", + "user_interaction_required": auth_type in ["user_credentials", "device_registration"], + } + + # Type-specific details + if auth_type == "user_credentials": + requirements.update( + { + "fields": ["username", "password"], + "optional_fields": ["client_id"], + "storage_key": "user_password", + } + ) + elif auth_type == "client_credentials": + requirements.update( + { + "fields": ["client_id", "client_secret"], + "storage_key": "client_credentials", + } + ) + elif auth_type == "network_based": + requirements.update( + { + "description": "Authenticates via your network provider", + "automatic": True, + } + ) + + return requirements + + # ===== AUTHENTICATION PROPERTIES AND METHODS ===== + + @property + def preferred_auth_type(self) -> str: + """Preferred authentication type (first in supported list).""" + types = self.supported_auth_types + return types[0] if types else "unknown" + + @property + def requires_stored_credentials(self) -> bool: + """True if provider needs credentials stored in settings.""" + credential_types = ["user_credentials", "client_credentials"] + return any(auth_type in credential_types for auth_type in self.supported_auth_types) + + @property + def requires_manifest_context(self) -> bool: + """True if provider needs to use same http manager to get manifest.""" + return False + + # ===== AUTHENTICATION PROPERTIES ===== + + def get_current_auth_type(self, context: AuthContext) -> str: + """ + Determine which auth type is currently active. + + Default implementation checks tokens/credentials. + Override for providers with complex auth logic. + + Args: + context: AuthContext for accessing tokens/credentials + + Returns: + Current active auth type + """ + return self._determine_current_auth_type_default(context) + + def _determine_current_auth_type_default(self, context: AuthContext) -> str: + """ + Default logic for determining current auth type. + Providers can override get_current_auth_type() directly instead. + """ + # 1. Check if provider requires stored credentials + if self.requires_stored_credentials: + credentials = context.get_credentials(self.provider_name, self.country) + if credentials: + # Map credential type to auth type + if hasattr(credentials, "credential_type"): + if credentials.credential_type == "user_password": + return "user_credentials" + elif credentials.credential_type == "client_credentials": + return "client_credentials" + + # 2. Check token auth level + primary_token = context.get_token( + self.provider_name, self.primary_token_scope, self.country + ) + if primary_token: + auth_level = primary_token.get("auth_level") + if auth_level == "user_authenticated": + return "user_credentials" + elif auth_level == "client_credentials": + return "client_credentials" + elif auth_level == "anonymous": + return "anonymous" + elif auth_level == "network_based": + return "network_based" + + # 3. Return first supported type as default + return self.preferred_auth_type + + # Token management properties (keep these) + @property + def primary_token_scope(self) -> Optional[str]: + """ + Primary token scope for this provider. + None = uses root-level token or no token needed. + + Returns: + Token scope string or None + """ + return None + + @property + def token_scopes(self) -> List[str]: + """ + All token scopes this provider uses. + + Returns: + List of token scope strings + """ + scope = self.primary_token_scope + return [scope] if scope else [] + + def get_auth_status(self, context: AuthContext) -> "AuthStatus": + """ + Get authentication status for this provider. + Uses AuthStatusBuilder by default. + + Override only for providers with special requirements. + + Args: + context: AuthContext with access to settings + + Returns: + AuthStatus object + """ + from ...providers.auth_builder import ( + AuthStatusBuilder, + ) # Import here to avoid circular imports + + return AuthStatusBuilder.for_provider(self, context) + + # Optional override methods for providers with special logic + def _calculate_auth_state(self, context: AuthContext): + """ + Override to provide custom auth state calculation. + Return None to use standard calculation. + + Returns: + AuthState or None + """ + return None + + def _calculate_readiness(self, context: AuthContext): + """ + Override to provide custom readiness calculation. + Return None to use standard calculation. + + Returns: + Tuple of (is_ready: bool, reason: str) or None + """ + return None + + def get_auth_details(self, context: AuthContext) -> Dict[str, Any]: + """ + Override to provide provider-specific auth details. + + Returns: + Dictionary with provider-specific information + """ + return {} \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/bookmarks.py b/lib/streaming_providers/base/provider_mixins/bookmarks.py new file mode 100644 index 0000000..fee1ca0 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/bookmarks.py @@ -0,0 +1,204 @@ +"""Provider bookmark (resume-position) mixin.""" + +from typing import Any, Dict, List, Optional + +from ..models.bookmark import Bookmark, ContentType + + +class ProviderBookmarksMixin: + @property + def implements_bookmarks(self) -> bool: + """ + True if this provider can store/retrieve playback positions. + + Return False (default) for providers that do not have bookmark / + continue-watching functionality. BookmarkOperations will skip + providers where this returns False when aggregating across all + providers. + + Providers that support bookmarks should override this to return True + and implement get_bookmarks(), update_bookmark(), and + delete_bookmark(). + """ + return False + + def get_bookmarks(self, **kwargs) -> List[Bookmark]: + """ + Return all bookmarks for the authenticated user. + + This method should fetch the user's continue-watching list or + playback positions from the provider's backend. + + Args: + **kwargs: Provider-specific filtering options (e.g., content_type, + limit, offset). + + Returns: + List of Bookmark objects, or [] if bookmarks are not supported + or none exist. + + Note: + The returned Bookmark objects should have their content_type field + properly set (LIVE, VOD, EVENT, RECORDING, etc.) so that the + client can correctly resolve the content. + + Default implementation returns an empty list. Override in provider + plugins that support bookmark storage. + """ + return [] + + def update_bookmark( + self, + content_id: str, + position_seconds: int, + content_type: ContentType, + duration_seconds: Optional[int] = None, + title: Optional[str] = None, + **kwargs, + ) -> Bookmark: + """ + Save or update a bookmark for specific content. + + This method is called automatically when playback stops or pauses. + The provider should store the position and associate it with the + authenticated user. + + Args: + content_id: The content being watched (channel ID, VOD ID, + etc.). + position_seconds: Where playback stopped in seconds from start. + 0 = not started / start of content. + -1 = explicitly marked as completed. + Content is also considered complete once position + reaches the model's COMPLETION_THRESHOLD + (>=95% by default). + content_type: Type of content being bookmarked. Required — + callers always know what they are bookmarking, + so this is never None. + duration_seconds: Total duration of the content (optional but + recommended for progress calculations). + title: Content title for caching (optional — provider + may ignore and use its own metadata store). + **kwargs: Provider-specific arguments (e.g., episode + number, season number, series ID). + + Returns: + The saved Bookmark object as confirmed by the provider. + + Raises: + RuntimeError: If the provider rejects the bookmark (e.g. user not + authenticated, content not accessible). + + Note: + At minimum the provider should persist content_id and + position_seconds so that get_bookmarks() can later return this + bookmark. + + Default implementation raises NotImplementedError so misconfigured + providers fail loudly rather than silently doing nothing. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement update_bookmark(). " + "Override this method to support bookmark storage." + ) + + def delete_bookmark(self, content_id: str, **kwargs) -> None: + """ + Delete a bookmark from the provider's backend. + + Called when: + - User manually removes a bookmark from "Continue Watching" + - Content is removed from the provider + - Cleanup of stale bookmarks + + Args: + content_id: The content identifier whose bookmark should be + removed. + **kwargs: Provider-specific arguments. + + Returns: + None on success. + + Raises: + KeyError: If no bookmark with this content_id exists. + RuntimeError: If the provider refuses deletion (e.g. permission + denied, backend error). + + Note: + Deleting a non-existent bookmark must raise KeyError rather than + silently succeeding, so that callers can distinguish "already + gone" from "successfully deleted". + + Default implementation raises NotImplementedError so misconfigured + providers fail loudly. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement delete_bookmark(). " + "Override this method to support bookmark deletion." + ) + + # ========================================================================= + # BOOKMARK HELPER METHODS (Optional overrides) + # ========================================================================= + + def batch_update_bookmarks( + self, updates: List[Dict[str, Any]], **kwargs + ) -> List[Bookmark]: + """ + Update multiple bookmarks in a single batch operation. + + Useful for synchronising local bookmark state with the provider + backend, or for bulk writes after a playback session. + + Args: + updates: List of dictionaries, each containing:: + + { + "content_id": str, + "position_seconds": int, + "content_type": ContentType, + "duration_seconds": Optional[int], + "title": Optional[str], + ... (other provider-specific kwargs) + } + + **kwargs: Provider-specific batch options passed to every + individual update_bookmark() call. + + Returns: + List of updated Bookmark objects in the same order as ``updates``. + If an individual update fails, the exception propagates and the + list contains only the bookmarks that succeeded before the + failure. + + Raises: + RuntimeError: If the batch operation fails partially or completely. + + Note: + The default implementation falls back to individual + update_bookmark() calls and does **not** mutate the dicts in + ``updates``. Override if your provider supports a native batch + endpoint for efficiency. + """ + results = [] + for raw in updates: + # Work on a copy so the caller's dicts are never mutated. + update = raw.copy() + content_id = update.pop("content_id") + position_seconds = update.pop("position_seconds") + content_type = update.pop("content_type") + duration_seconds = update.pop("duration_seconds", None) + title = update.pop("title", None) + + result = self.update_bookmark( + content_id=content_id, + position_seconds=position_seconds, + content_type=content_type, + duration_seconds=duration_seconds, + title=title, + **update, + **kwargs, + ) + results.append(result) + + return results \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/catchup.py b/lib/streaming_providers/base/provider_mixins/catchup.py new file mode 100644 index 0000000..a89e355 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/catchup.py @@ -0,0 +1,284 @@ +"""Provider catchup (start-over / replay) mixin.""" + +import time +from datetime import datetime +from typing import Dict, List, Optional, Tuple + +from ..models import DRMConfig + + +class ProviderCatchupMixin: + @property + def catchup_window(self) -> int: + """ + Return the catchup window in HOURS for this provider. + + Returns: + int: Number of hours of catchup available (0 = no catchup support) + """ + return 0 + + @property + def supports_catchup(self) -> bool: + """ + Check if provider supports catchup/timeshift functionality. + + Returns: + bool: True if catchup is supported + """ + return self.catchup_window > 0 + + def get_catchup_manifest( + self, + content_id: str, + start_time: int, + end_time: int, + epg_id: Optional[str] = None, + **kwargs, + ) -> Optional[str]: + """ + Get manifest URL for catchup/timeshift content. + + Args: + content_id: Channel identifier + start_time: Start time as Unix timestamp + end_time: End time as Unix timestamp + epg_id: Optional EPG event ID (might be needed by some providers) + **kwargs: Additional provider-specific parameters + + Returns: + Manifest URL for catchup content, or None if not supported + + Default implementation raises NotImplementedError. + Override in subclass to implement provider-specific catchup logic. + + Do NOT fall back to self.get_manifest() here — returning the live + manifest URL as a catchup manifest will cause the DRM pipeline to + extract PSSH from the live stream, which may differ from the catchup + stream's encryption context. + """ + raise NotImplementedError( + f"{self.__class__.__name__}.get_catchup_manifest() is not implemented." + ) + + def get_catchup_manifest_headers( + self, + content_id: str, + start_time: int, + end_time: int, + epg_id: Optional[str] = None, + **kwargs, + ) -> Dict[str, str]: + """ + Return HTTP headers for the catchup manifest request. + + Default implementation delegates to get_manifest_headers() since many + providers use the same auth headers for live and catchup manifests. + Override when catchup requires different headers (e.g. extra tokens). + """ + return self.get_manifest_headers(content_id, **kwargs) + + def get_catchup_manifest_with_headers( + self, + content_id: str, + start_time: int, + end_time: int, + epg_id: Optional[str] = None, + **kwargs, + ) -> Tuple[Optional[str], Dict[str, str]]: + """ + Convenience method returning (catchup_manifest_url, headers). + + This is the single entry point used by CatchupOperations before + calling into the DRM pipeline, mirroring the role that + get_manifest_with_headers() plays for live content. + + Providers should override get_catchup_manifest() (and optionally + get_catchup_manifest_headers()) rather than this method directly. + + Raises: + NotImplementedError: propagated from get_catchup_manifest() if the + provider has not implemented catchup manifest resolution. + """ + url = self.get_catchup_manifest( + content_id=content_id, + start_time=start_time, + end_time=end_time, + epg_id=epg_id, + **kwargs, + ) + headers = self.get_catchup_manifest_headers( + content_id=content_id, + start_time=start_time, + end_time=end_time, + epg_id=epg_id, + **kwargs, + ) + return url, headers + + def get_catchup_drm( + self, + content_id: str, + start_time: int, + end_time: int, + epg_id: Optional[str] = None, + drm_variant: Optional[str] = None, + **kwargs, + ) -> List[DRMConfig]: + """ + Get DRM configurations for catchup content. + + Args: + content_id: Channel identifier + start_time: Start time as Unix timestamp + end_time: End time as Unix timestamp + epg_id: Optional EPG event ID (might be needed for DRM licensing) + drm_variant: Optional DRM variant ('auto', 'software', 'hardware') + **kwargs: Additional provider-specific parameters + + Returns: + List of DRM configurations for catchup content + + Default implementation raises NotImplementedError so that the DRM + pipeline falls through to PSSH extraction from the catchup manifest. + + Override in subclass when catchup requires a *different* DRM + configuration from live (e.g. a different license URL, extra request + headers, or a static ClearKey set). If catchup uses exactly the same + DRM as live, implement as: + + def get_catchup_drm(self, content_id, start_time, end_time, + epg_id=None, drm_variant=None, **kwargs): + return self.get_drm(content_id, drm_variant=drm_variant, **kwargs) + + Do NOT call super().get_drm() silently — that would make the pipeline + think Phase 2 produced valid configs from the live stream context, + which is wrong when the catchup manifest has different encryption. + """ + raise NotImplementedError( + f"{self.__class__.__name__}.get_catchup_drm() is not implemented. " + "The DRM pipeline will extract PSSH from the catchup manifest directly. " + "Override this method only if catchup requires a custom DRM configuration." + ) + + # ============================================================================ + # CATCHUP HELPER METHODS + # ============================================================================ + + def get_catchup_window_for_channel(self, content_id: str) -> int: + """ + Get catchup window for a specific channel in HOURS. + + Args: + content_id: Channel identifier + + Returns: + int: Catchup window in hours for this channel + """ + return self.catchup_window + + def validate_catchup_request( + self, start_time: int, end_time: int + ) -> tuple[bool, Optional[str]]: + """ + Validate a catchup request against provider's capabilities. + + Args: + start_time: Start time as Unix timestamp + end_time: End time as Unix timestamp + + Returns: + Tuple of (is_valid, error_message) + """ + if not self.supports_catchup: + return False, f"Provider '{self.provider_name}' does not support catchup" + + if start_time >= end_time: + return False, "Invalid time range: start_time must be before end_time" + + now = int(time.time()) + if start_time > now: + return False, "Cannot request future content" + + # CHANGE FROM DAYS TO HOURS HERE + max_age_seconds = self.catchup_window * 3600 # hours to seconds + content_age = now - start_time + + if content_age > max_age_seconds: + hours_ago = content_age // 3600 + return False, ( + f"Content is outside catchup window " + f"(requested: {hours_ago} hours ago, " + f"max: {self.catchup_window} hours)" + ) + + return True, None + + def format_catchup_time_params( + self, start_time: int, end_time: int, format_type: str = "iso" + ) -> Dict[str, str]: + """ + Format time parameters for provider-specific API calls. + + Different providers expect different time formats in their APIs. + This helper converts Unix timestamps to various formats. + + Args: + start_time: Start time as Unix timestamp + end_time: End time as Unix timestamp + format_type: Format type ('iso', 'unix', 'millis', 'custom') + + Returns: + Dictionary with formatted time parameters + + Override in subclass for provider-specific formatting. + """ + if format_type == "iso": + # ISO 8601 format + start_dt = datetime.fromtimestamp(start_time) + end_dt = datetime.fromtimestamp(end_time) + return {"start": start_dt.isoformat(), "end": end_dt.isoformat()} + elif format_type == "unix": + # Unix timestamps (seconds) + return {"start": str(start_time), "end": str(end_time)} + elif format_type == "millis": + # Milliseconds since epoch + return {"start": str(start_time * 1000), "end": str(end_time * 1000)} + else: + # Default to unix + return {"start": str(start_time), "end": str(end_time)} + + def build_catchup_manifest_url( + self, base_url: str, start_time: int, end_time: int, url_format: str = "query" + ) -> str: + """ + Build catchup manifest URL with time parameters. + + Helper method to construct manifest URLs with time parameters + in various formats that different providers use. + + Args: + base_url: Base manifest URL + start_time: Start time as Unix timestamp + end_time: End time as Unix timestamp + url_format: Format ('query', 'path', 'fragment') + + Returns: + Complete manifest URL with time parameters + + Override in subclass for provider-specific URL construction. + """ + if url_format == "query": + # Add as query parameters + separator = "&" if "?" in base_url else "?" + return f"{base_url}{separator}start={start_time}&end={end_time}" + elif url_format == "path": + # Add to path (e.g., /manifest/start/end.mpd) + return f"{base_url}/{start_time}/{end_time}" + elif url_format == "fragment": + # Add as URL fragment (e.g., manifest.mpd#t=start,end) + return f"{base_url}#t={start_time},{end_time}" + else: + # Default to query parameters + separator = "&" if "?" in base_url else "?" + return f"{base_url}{separator}start={start_time}&end={end_time}" \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/epg.py b/lib/streaming_providers/base/provider_mixins/epg.py new file mode 100644 index 0000000..488590c --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/epg.py @@ -0,0 +1,83 @@ +"""Provider EPG (electronic program guide) mixin.""" + +from datetime import datetime +from typing import Dict, List, Optional, Tuple + +from ..models.epg_models import EPGEntry, EPGProgramDetails + + +class ProviderEpgMixin: + @property + def epg_window(self) -> Tuple[int, int]: + """ + Return the EPG window as (past_days, future_days). + + Returns: + Tuple[int, int]: (past_days, future_days) + (0, 0) means no EPG support + + MUST be overridden by providers. + """ + return 0, 0 + + @property + def implements_epg(self) -> bool: + """Check if provider implements EPG.""" + return self.epg_window != (0, 0) + + # ============================================================================ + # EPG METHODS - Optional with sensible defaults + # ============================================================================ + + def get_epg( + self, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + country: Optional[str] = None, + **kwargs, + ) -> List["EPGEntry"]: + """ + Get EPG data for a specific channel. + + Override if provider supports per-channel EPG. + Default returns empty list (no EPG). + """ + return [] + + def get_epg_grid( + self, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + channel_ids: Optional[List[str]] = None, + country: Optional[str] = None, + **kwargs, + ) -> Dict[str, List["EPGEntry"]]: + """ + Get EPG data for multiple channels in one operation. + + Override if provider supports batch EPG. + Default returns empty dict (no batch EPG). + + Note: If provider only supports per-channel EPG, + implement get_epg() and leave this as default. + """ + return {} + + def get_program_details(self, program_id: str, **kwargs) -> Optional["EPGProgramDetails"]: + """ + Get detailed metadata for a single program. + + Override if provider supports program details. + Default returns None (no details). + """ + return None + + def get_epg_xmltv(self, country: Optional[str] = None, **kwargs) -> Optional[str]: + """ + Get complete EPG data in XMLTV format. + + Override if provider supports XMLTV export. + Default returns None (no XMLTV). + """ + return None \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/favorites.py b/lib/streaming_providers/base/provider_mixins/favorites.py new file mode 100644 index 0000000..a26d998 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/favorites.py @@ -0,0 +1,65 @@ +"""Provider favorites mixin.""" + +from typing import List, Optional + +from ..models.favorite import Favorite, FavoriteType + + +class ProviderFavoritesMixin: + @property + def implements_favorites(self) -> bool: + """ + True if this provider can store/retrieve user favorites. + + Default False. Override in providers that support favorites. + """ + return False + + def get_favorites(self, **kwargs) -> List["Favorite"]: + """ + Return all favorites for the authenticated user. + + Returns: + List of Favorite objects, or [] if not supported or none exist. + """ + return [] + + def add_favorite( + self, + content_id: str, + favorite_type: FavoriteType, + title: Optional[str] = None, + **kwargs, + ) -> "Favorite": + """ + Add a content to user's favorites. + + Args: + content_id: Content identifier. + favorite_type: Type of content (PROGRAM, CLIP, LIVE, EVENT). + title: Content title (optional for caching). + + Returns: + The saved Favorite object. + + Raises: + RuntimeError: If the provider rejects the favorite. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement add_favorite()." + ) + + def remove_favorite(self, content_id: str, **kwargs) -> None: + """ + Remove a content from user's favorites. + + Args: + content_id: Content identifier to remove. + + Raises: + KeyError: If no favorite with this content_id exists. + RuntimeError: If the provider refuses deletion. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement remove_favorite()." + ) \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/http.py b/lib/streaming_providers/base/provider_mixins/http.py new file mode 100644 index 0000000..34f533d --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/http.py @@ -0,0 +1,152 @@ +"""Provider HTTP-manager setup mixin: proxy resolution and HTTPManager creation.""" + +from typing import TYPE_CHECKING, Dict, Optional + +from ..models.proxy_models import ProxyConfig +from ..network import HTTPManager, HTTPManagerFactory +from ..utils.logger import logger + + +class ProviderHttpMixin: + if TYPE_CHECKING: + # Provided by StreamingProvider.__init__ once this is composed + # into the full class — declared here only for type checkers. + provider_name: str + country: str + _http_manager: Optional[HTTPManager] + + @property + def http_manager(self) -> Optional[HTTPManager]: + """Return the provider's HTTP manager instance""" + return self._http_manager + + @http_manager.setter + def http_manager(self, value: HTTPManager): + """Set the provider's HTTP manager instance""" + self._http_manager = value + + def _setup_http_manager( + self, + provider_name: str, + proxy_config: Optional[ProxyConfig] = None, + proxy_url: Optional[str] = None, + config_dir: Optional[str] = None, + country: Optional[str] = None, + user_agent: Optional[str] = None, + timeout: Optional[int] = None, + max_retries: Optional[int] = None, + **kwargs, + ) -> HTTPManager: + """Standard HTTP manager setup for providers with intelligent proxy resolution""" + if country is None: + country = self.country + + resolved_proxy = self._resolve_proxy_config( + proxy_config=proxy_config, + proxy_url=proxy_url, + config_dir=config_dir, + provider_name=provider_name, + country=country, + ) + + manager_kwargs = {} + if user_agent: + manager_kwargs["user_agent"] = user_agent + if timeout: + manager_kwargs["timeout"] = timeout + if max_retries: + manager_kwargs["max_retries"] = max_retries + manager_kwargs.update(kwargs) + + http_manager = HTTPManagerFactory.create_for_provider( + provider_name=provider_name, proxy_config=resolved_proxy, **manager_kwargs + ) + + self._log_http_manager_setup(provider_name, resolved_proxy, manager_kwargs) + return http_manager + + @staticmethod + def _resolve_proxy_config( + proxy_config: Optional[ProxyConfig], + proxy_url: Optional[str], + config_dir: Optional[str], + provider_name: str, + country: str, + ) -> Optional[ProxyConfig]: + """Resolve proxy configuration from multiple sources with priority""" + if proxy_config is not None: + logger.debug(f"{provider_name}: Using directly provided proxy configuration") + return proxy_config + + if proxy_url: + try: + logger.debug(f"{provider_name}: Creating proxy config from URL") + return ProxyConfig.from_url(proxy_url) + except Exception as e: + logger.warning(f"{provider_name}: Failed to parse proxy URL '{proxy_url}': {e}") + + try: + from ..network import ProxyConfigManager + + proxy_mgr = ProxyConfigManager(config_dir) + managed_proxy = proxy_mgr.get_proxy_config(provider_name, country) + + if managed_proxy: + logger.debug(f"{provider_name}: Using proxy from ProxyConfigManager") + return managed_proxy + else: + logger.debug(f"{provider_name}: No proxy configuration found in ProxyConfigManager") + + except Exception as e: + logger.warning(f"{provider_name}: Could not load proxy from ProxyConfigManager: {e}") + + logger.debug(f"{provider_name}: No proxy configuration available") + return None + + @staticmethod + def _log_http_manager_setup( + provider_name: str, proxy_config: Optional[ProxyConfig], manager_kwargs: Dict + ) -> None: + """Log HTTP manager setup information""" + info_parts = [f"HTTP manager initialized for '{provider_name}'"] + + if proxy_config: + proxy_type = proxy_config.proxy_type.value if proxy_config.proxy_type else "http" + proxy_host = f"{proxy_config.host}:{proxy_config.port}" + has_auth = "authenticated" if proxy_config.auth else "no-auth" + info_parts.append(f"proxy: {proxy_type}://{proxy_host} ({has_auth})") + else: + info_parts.append("proxy: none") + + if "user_agent" in manager_kwargs: + ua_preview = ( + manager_kwargs["user_agent"][:50] + "..." + if len(manager_kwargs["user_agent"]) > 50 + else manager_kwargs["user_agent"] + ) + info_parts.append(f"user-agent: {ua_preview}") + + if "timeout" in manager_kwargs: + info_parts.append(f"timeout: {manager_kwargs['timeout']}s") + + if "max_retries" in manager_kwargs: + info_parts.append(f"retries: {manager_kwargs['max_retries']}") + + logger.info(f"{provider_name}: {', '.join(info_parts)}") + + def _share_http_manager_with_authenticator( + self, authenticator, http_manager: Optional[HTTPManager] = None + ) -> HTTPManager: + """Share HTTP manager with authenticator for consistency""" + if http_manager is None: + http_manager = self.http_manager + + if http_manager and hasattr(authenticator, "http_manager"): + if authenticator.http_manager is None: + logger.debug(f"{self.provider_name}: Sharing HTTP manager with authenticator") + authenticator.http_manager = http_manager + else: + logger.debug(f"{self.provider_name}: Using authenticator's existing HTTP manager") + http_manager = authenticator.http_manager + + return http_manager \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/metadata.py b/lib/streaming_providers/base/provider_mixins/metadata.py new file mode 100644 index 0000000..67910ac --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/metadata.py @@ -0,0 +1,170 @@ +"""Provider static-metadata mixin: labels, logos, auth-type lists, country handling. + +Split out of provider.py. Reads its data from the ClassVar attributes +(PROVIDER_LABEL, SUPPORTED_AUTH_TYPES, PROVIDER_LOGO, SUPPORTED_COUNTRIES) +that remain declared directly on StreamingProvider, since those describe the +concrete provider class itself, not any one capability area. +""" + +from typing import TYPE_CHECKING, Any, Dict, List + + +class ProviderMetadataMixin: + if TYPE_CHECKING: + # Provided by StreamingProvider.__init__ once this is composed + # into the full class — declared here only for type checkers. + country: str + + @classmethod + def get_static_label(cls, country: str = None) -> str: + """ + Get provider label without instantiation. + + Args: + country: Optional country code for country-specific labels + + Returns: + Provider label string + """ + base_label = cls.PROVIDER_LABEL or cls.__name__.replace("Provider", "") + + if country: + # Format country code + country_upper = country.upper() + + # Special handling for common cases + if country_upper == "DE": + return f"{base_label} Germany" + elif country_upper == "AT": + return f"{base_label} Austria" + elif country_upper == "CH": + return f"{base_label} Switzerland" + else: + return f"{base_label} ({country_upper})" + + return base_label + + @classmethod + def get_static_auth_types(cls) -> List[str]: + """ + Get supported authentication types without instantiation. + + Returns: + List of supported auth type strings + """ + return cls.SUPPORTED_AUTH_TYPES.copy() + + @classmethod + def get_static_logo(cls, country: str = None) -> str: + """ + Get provider logo URL without instantiation. + + Args: + country: Optional country code for country-specific logos + + Returns: + Logo URL string + """ + return cls.PROVIDER_LOGO + + @classmethod + def get_static_supported_countries(cls) -> List[str]: + """ + Get supported countries without instantiation. + + Returns: + List of ISO country codes + """ + return cls.SUPPORTED_COUNTRIES.copy() + + @classmethod + def get_all_possible_instances(cls) -> List[Dict[str, Any]]: + """ + Get metadata for all possible instances of this provider. + + Returns: + List of instance metadata dictionaries + """ + instances = [] + + if cls.supports_multiple_countries(): + for country in cls.SUPPORTED_COUNTRIES: + instances.append( + { + "plugin": cls.__name__.lower().replace("provider", ""), + "country": country.upper(), + "label": cls.get_static_label(country), + "requires_country_suffix": True, + } + ) + else: + # Single-country provider + instances.append( + { + "plugin": cls.__name__.lower().replace("provider", ""), + "country": "DE", # Default country for single-country providers + "label": cls.get_static_label(), + "requires_country_suffix": False, + } + ) + + return instances + + @property + def provider_label(self) -> str: + """Return the provider label (e.g., 'JOYN', 'ZDF', 'RTL+')""" + # Use static method with instance's country + return self.get_static_label(self.country) + + @property + def provider_logo(self) -> str: + """Return the provider logo URL""" + return self.get_static_logo() + + @property + def supported_auth_types(self) -> List[str]: + """List of authentication types this provider supports.""" + return self.get_static_auth_types() + + @property + def uses_dynamic_manifests(self) -> bool: + """Return True if provider uses truly dynamic manifests""" + return False + + @classmethod + def get_supported_countries(cls) -> List[str]: + """ + Get list of countries supported by this provider. + + Returns: + List of ISO country codes (e.g., ['de', 'at', 'ch']) + Empty list means single-country provider using default country + """ + return cls.SUPPORTED_COUNTRIES.copy() + + @classmethod + def supports_multiple_countries(cls) -> bool: + """ + Check if this provider supports multiple countries. + + Returns: + True if provider supports country-specific instances + """ + return len(cls.SUPPORTED_COUNTRIES) > 1 + + @classmethod + def validate_country(cls, country: str) -> bool: + """ + Validate if a country is supported by this provider. + + Args: + country: ISO country code to validate + + Returns: + True if country is supported or provider is single-country + """ + if not cls.supports_multiple_countries(): + # Single-country providers accept any country (or ignore it) + return True + + return country.lower() in [c.lower() for c in cls.SUPPORTED_COUNTRIES] \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/recordings.py b/lib/streaming_providers/base/provider_mixins/recordings.py new file mode 100644 index 0000000..39b6bb9 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/recordings.py @@ -0,0 +1,61 @@ +"""Provider recording (cloud/network PVR) mixin.""" + +from typing import List + +from ..models.recording import Recording + + +class ProviderRecordingsMixin: + @property + def implements_recordings(self) -> bool: + """ + True if this provider can return recorded content. + + Return False (default) for providers that only offer live or VOD content. + RecordingOperations skips providers where this returns False when + aggregating across all providers. + """ + return False + + def get_recordings( + self, + include_deleted: bool = False, + **kwargs, + ) -> List[Recording]: + """ + Return recordings available for the authenticated user. + + Args: + include_deleted: If True, also return recordings marked as deleted + (useful for a trash/recycle-bin view). + **kwargs: Provider-specific filtering options. + + Returns: + List of Recording objects, or [] if recordings are not supported. + """ + return [] + + def delete_recording(self, recording_id: str, **kwargs) -> None: + """ + Permanently delete a recording on the provider's backend. + + Args: + recording_id: The recording to delete (== content_id on the model). + **kwargs: Provider-specific arguments. + + Returns: + None on success. + + Raises: + KeyError: If no recording with this ID exists. + RuntimeError: If the provider refuses the deletion (e.g. the + recording is currently being captured, or the user + lacks permission). + + Default implementation raises NotImplementedError so misconfigured + providers fail loudly rather than silently doing nothing. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement delete_recording(). " + "Override this method to support recording deletion." + ) \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/subscriptions.py b/lib/streaming_providers/base/provider_mixins/subscriptions.py new file mode 100644 index 0000000..6e68d67 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/subscriptions.py @@ -0,0 +1,111 @@ +"""Provider subscription/entitlement mixin.""" + +from typing import TYPE_CHECKING, List, Optional + +from ..models import StreamingChannel +from ..models.subscription import SubscriptionPackage, UserSubscription + + +class ProviderSubscriptionsMixin: + if TYPE_CHECKING: + # Provided by StreamingProvider once this is composed into the + # full class — declared here only for type checkers. + def get_channels(self, **kwargs) -> List[StreamingChannel]: ... + + def get_subscription_status(self, **kwargs) -> Optional[UserSubscription]: + """ + Get user's subscription status for this provider. + + Returns: + UserSubscription object with subscription details, or + None if provider doesn't support subscription queries or + subscription info isn't available. + + Default implementation returns None (subscription not supported). + Override in provider plugins that support subscription checking. + + Example usage in providers: + # Query provider API for user entitlements + # Parse response into SubscriptionPackage objects + # Return UserSubscription with packages and accessible channels + """ + return None + + def get_subscribed_channels(self, **kwargs) -> List[StreamingChannel]: + """ + Get channels the current user is subscribed to. + + This method: + 1. Gets subscription status (if supported) + 2. Filters all channels based on accessible channel IDs + 3. Returns filtered list or all channels as fallback + + Returns: + List of StreamingChannel objects that the user can access + + Note: + Override get_subscription_status() in provider plugins + to enable subscription filtering. + """ + # Get all channels first + all_channels = self.get_channels(**kwargs) + + # Try to get subscription status + subscription = self.get_subscription_status(**kwargs) + + # If no subscription info or not active, return all channels + if not subscription or not subscription.active: + return all_channels + + # Filter channels based on accessible channel IDs + if subscription.accessible_channel_ids: + return [ + channel + for channel in all_channels + if channel.channel_id in subscription.accessible_channel_ids + ] + + # No filtering possible, return all channels + return all_channels + + def get_available_packages(self, **kwargs) -> List[SubscriptionPackage]: + """ + Get all subscription packages available from this provider. + + Useful for: + - Displaying upgrade options in UI + - Showing package comparison + - Subscription management interface + + Returns: + List of available SubscriptionPackage objects + Empty list if not implemented or no packages available + + Default implementation returns empty list. + Override in provider plugins that have package information. + """ + return [] + + def is_channel_accessible(self, channel_id: str, **kwargs) -> bool: + """ + Check if a specific channel is accessible with current subscription. + + Args: + channel_id: ID of the channel to check + **kwargs: Additional arguments passed to get_subscription_status() + + Returns: + True if channel is accessible, False otherwise. + Returns True if subscription checking is not supported. + + Note: + This is a convenience method for quick checks. + """ + subscription = self.get_subscription_status(**kwargs) + + # If no subscription info, assume accessible (backward compatibility) + if not subscription or not subscription.active: + return True + + # Check if channel is in accessible set + return subscription.can_access_channel(channel_id) \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/timers.py b/lib/streaming_providers/base/provider_mixins/timers.py new file mode 100644 index 0000000..6f20733 --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/timers.py @@ -0,0 +1,116 @@ +"""Provider timer (scheduled recording) mixin.""" + +from typing import List + +from ..models.timer import Timer +from ..models.timer_type import TimerType + + +class ProviderTimersMixin: + @property + def implements_timers(self) -> bool: + """ + True if this provider supports scheduled recording timers. + + Return False (default) for providers that only offer live or VOD content + and have no PVR/timer backend. TimerOperations skips providers where + this returns False when aggregating across all providers. + """ + return False + + def get_timer_types(self) -> List["TimerType"]: + """ + Return the timer types this provider supports. + + Providers that support timers MUST override this method and return at + least one TimerType so that clients know what fields to present when + creating a timer. + + Returns: + List of TimerType objects, or [] if timers are not supported. + """ + return [] + + def get_timers(self, **kwargs) -> List["Timer"]: + """ + Return all timers (scheduled recordings) for the authenticated user. + + Args: + **kwargs: Provider-specific filtering options. + + Returns: + List of Timer objects, or [] if timers are not supported. + """ + return [] + + def add_timer(self, timer: "Timer", **kwargs) -> "Timer": + """ + Schedule a new timer on the provider's backend. + + Args: + timer: Timer to create. timer.client_index is ignored — the + provider allocates and sets it on the returned object. + **kwargs: Provider-specific arguments. + + Returns: + The saved Timer with client_index populated by the provider. + + Raises: + RuntimeError: If the provider rejects the timer (e.g. scheduling + conflict, unsupported timer type, insufficient + permissions). + + Default raises NotImplementedError so misconfigured providers fail + loudly rather than silently doing nothing. + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement add_timer(). " + "Override this method to support timer creation." + ) + + def update_timer(self, timer: "Timer", **kwargs) -> "Timer": + """ + Update an existing timer on the provider's backend. + + Args: + timer: Timer with updated fields. timer.client_index identifies + the record to modify. + **kwargs: Provider-specific arguments. + + Returns: + The updated Timer as confirmed by the provider. + + Raises: + KeyError: If no timer with that client_index exists. + RuntimeError: If the provider refuses the update (e.g. the timer + is currently recording). + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement update_timer(). " + "Override this method to support timer updates." + ) + + def delete_timer( + self, client_index: int, force_delete: bool = False, **kwargs + ) -> None: + """ + Delete a timer on the provider's backend. + + Args: + client_index: Timer identifier to delete. + force_delete: If True and the timer is currently recording, abort + the ongoing capture before deleting. + **kwargs: Provider-specific arguments. + + Returns: + None on success. + + Raises: + KeyError: If no timer with that client_index exists. + RuntimeError: If the provider refuses deletion (e.g. recording in + progress and force_delete is False). + """ + raise NotImplementedError( + f"{self.__class__.__name__} does not implement delete_timer(). " + "Override this method to support timer deletion." + ) \ No newline at end of file diff --git a/lib/streaming_providers/base/provider_mixins/vod.py b/lib/streaming_providers/base/provider_mixins/vod.py new file mode 100644 index 0000000..58edb9e --- /dev/null +++ b/lib/streaming_providers/base/provider_mixins/vod.py @@ -0,0 +1,56 @@ +"""Provider video-on-demand (VOD) mixin.""" + +from typing import List, Optional + + +class ProviderVodMixin: + @property + def implements_vod(self) -> bool: + """ + Indicates whether this provider has a browsable VOD catalogue. + + Return False (and let get_vod_category return []) for providers that + only offer live channels or events. VodOperations will skip providers + where this returns False when aggregating across all providers. + """ + return False + + def get_vod_category(self, content_id: str = "", **kwargs) -> List: + """ + Return the children of a VOD tree node. + + Args: + content_id: Opaque node identifier returned by a previous + get_vod_category call. Empty string → root level. + Providers define their own ID format; the caller + treats it as an opaque token and never parses it. + + Returns: + Mixed list of VodCategory and VodItem objects. + Return [] if the node has no children or VOD is not supported. + """ + return [] + + def search_vod( + self, + query: str, + cursor: Optional[str] = None, + page_size: int = 24, + **kwargs, + ) -> List: + """ + Search the VOD catalogue for items matching query. + + Args: + query: Free-text search string entered by the user. + cursor: Opaque continuation token from a previous response's + next_cursor field. None → first page. + page_size: Hint for how many entries to return per page. + Providers may ignore or clamp this value. + + Returns: + Mixed list of VodCategory and VodItem objects, or a paged dict + with the same shape as get_vod_category (entries, next_cursor, + total). Return [] if search is not supported. + """ + return [] \ No newline at end of file