diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index 985b3a0..23a8e8b 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -14,6 +14,7 @@ from ...base.models.streaming_channel import StreamingChannel from ...base.network import HTTPManagerFactory, ProxyConfigManager from ...base.models.proxy_models import ProxyConfig from ...base.utils.logger import logger +from ...base.models.auth import AuthState from .models import Magenta2Channel, Magenta2PlaybackRestrictedException from .auth import Magenta2Authenticator, Magenta2Credentials, Magenta2UserCredentials from .discovery import DiscoveryService @@ -1539,6 +1540,137 @@ class Magenta2Provider(StreamingProvider): logger.error(f"Error getting EPG for channel {channel_id}: {e}") return [] + def _calculate_auth_state(self, context) -> 'AuthState': + """ + Custom auth state calculation for Magenta2. + Checks persona token with its special structure. + """ + from ...base.models.auth import AuthState + import time + + # Get persona token data + persona_token = context.get_token(self.provider_name, 'persona', self.country) + + if not persona_token: + logger.debug("No persona token found") + return AuthState.NOT_AUTHENTICATED + + # Validate persona token structure + if not isinstance(persona_token, dict) or 'persona_token' not in persona_token: + logger.warning("Invalid persona token structure") + return AuthState.NOT_AUTHENTICATED + + # Check if persona token is expired + if 'expires_at' in persona_token: + current_time = time.time() + expires_at = persona_token['expires_at'] + + # Add 5-minute buffer + if current_time >= (expires_at - 300): + logger.debug(f"Persona token expired (expires_at: {expires_at}, now: {current_time})") + return AuthState.EXPIRED + + # Token exists and is valid + logger.debug("Persona token is valid") + return AuthState.AUTHENTICATED + + def _calculate_readiness(self, context) -> Tuple[bool, str]: + """ + Custom readiness calculation for Magenta2. + Ready if we have a valid persona token. + """ + import time + + # Get persona token data + persona_token = context.get_token(self.provider_name, 'persona', self.country) + + if not persona_token: + return False, "No persona token available" + + # Validate structure + if not isinstance(persona_token, dict) or 'persona_token' not in persona_token: + return False, "Invalid persona token structure" + + # Check expiration + if 'expires_at' in persona_token: + current_time = time.time() + expires_at = persona_token['expires_at'] + + # Add 5-minute buffer + if current_time >= (expires_at - 300): + # Token expired but might be refreshable + # Check if we have yo_digital token with refresh capability + yo_token = context.get_token(self.provider_name, 'yo_digital', self.country) + if yo_token and 'refresh_token' in yo_token: + # Check if yo_digital refresh token is still valid + if 'refresh_token_expires_in' in yo_token and 'refresh_token_issued_at' in yo_token: + refresh_expires_at = yo_token['refresh_token_issued_at'] + yo_token['refresh_token_expires_in'] + if current_time < (refresh_expires_at - 300): + return True, "Persona token expired but can be refreshed via yo_digital" + + return False, f"Persona token expired (expired at {time.ctime(expires_at)})" + + # Valid persona token + return True, "Has valid persona token" + + def get_auth_details(self, context) -> Dict[str, Any]: + """ + Provide Magenta2-specific auth details. + Shows status of all token scopes. + """ + import time + + details = {} + + # Check all token scopes + for scope in self.token_scopes: + token = context.get_token(self.provider_name, scope, self.country) + + if not token: + details[scope] = {"available": False} + continue + + scope_info = {"available": True} + + # Handle persona token specially + if scope == 'persona': + if 'expires_at' in token: + current_time = time.time() + expires_at = token['expires_at'] + scope_info['expires_at'] = expires_at + scope_info['is_expired'] = current_time >= expires_at + scope_info['time_remaining'] = int(max(0, expires_at - current_time)) + + if 'composed_at' in token: + scope_info['composed_at'] = token['composed_at'] + + # Handle yo_digital token + elif scope == 'yo_digital': + if 'access_token_expires_in' in token and 'access_token_issued_at' in token: + current_time = time.time() + expires_at = token['access_token_issued_at'] + token['access_token_expires_in'] + scope_info['access_token_expires_at'] = expires_at + scope_info['access_token_is_expired'] = current_time >= expires_at + + if 'refresh_token_expires_in' in token and 'refresh_token_issued_at' in token: + current_time = time.time() + refresh_expires_at = token['refresh_token_issued_at'] + token['refresh_token_expires_in'] + scope_info['refresh_token_expires_at'] = refresh_expires_at + scope_info['refresh_token_is_expired'] = current_time >= refresh_expires_at + scope_info['has_refresh_token'] = True + + # Standard token handling (tvhubs, taa) + else: + if 'expires_in' in token and 'issued_at' in token: + current_time = time.time() + expires_at = token['issued_at'] + token['expires_in'] + scope_info['expires_at'] = expires_at + scope_info['is_expired'] = current_time >= expires_at + + details[scope] = scope_info + + return details + def debug_authentication(self) -> Dict[str, Any]: """ Debug authentication state and token flow