diff --git a/lib/streaming_providers/base/models/auth.py b/lib/streaming_providers/base/models/auth.py index 18ef7e0..91f1688 100644 --- a/lib/streaming_providers/base/models/auth.py +++ b/lib/streaming_providers/base/models/auth.py @@ -1,23 +1,20 @@ # streaming_providers/base/models/auth.py -from dataclasses import dataclass, field -from typing import Dict, Any, Optional +from typing import Optional, Dict, Any from enum import Enum -import time +from dataclasses import dataclass class AuthState(Enum): - """Standardized authentication states""" + """Authentication state""" + NOT_APPLICABLE = "not_applicable" NOT_AUTHENTICATED = "not_authenticated" AUTHENTICATED = "authenticated" EXPIRED = "expired" - PENDING = "pending" - ERROR = "error" - NOT_APPLICABLE = "not_applicable" @dataclass class TokenInfo: - """Standardized token information""" + """Information about a specific token scope""" scope: str has_token: bool is_valid: bool @@ -25,93 +22,89 @@ class TokenInfo: has_refresh_token: bool = False auth_level: Optional[str] = None - def to_dict(self) -> Dict[str, Any]: - return { - 'scope': self.scope, - 'has_token': self.has_token, - 'is_valid': self.is_valid, - 'expires_at': self.expires_at, - 'has_refresh_token': self.has_refresh_token, - 'auth_level': self.auth_level - } - @dataclass class AuthStatus: """Complete authentication status for a provider""" - # Core identification provider_name: str provider_label: str - country: str - - # Authentication + country: Optional[str] auth_type: str auth_state: AuthState - - # Readiness is_ready: bool - readiness_reason: Optional[str] = None + readiness_reason: str + requires_stored_credentials: bool + has_credentials: bool + credentials_type: Optional[str] + has_valid_token: bool + primary_token_scope: Optional[str] + token_scopes: Dict[str, TokenInfo] + last_authentication: Optional[float] + provider_specific: Dict[str, Any] - # Tokens - primary_token_scope: Optional[str] = None - token_scopes: Dict[str, TokenInfo] = field(default_factory=dict) - has_valid_token: bool = False - - # Credentials - requires_stored_credentials: bool = True - has_credentials: bool = False - credentials_type: Optional[str] = None - - # Metadata - timestamp: float = field(default_factory=time.time) - last_authentication: Optional[float] = None - provider_specific: Dict[str, Any] = field(default_factory=dict) + # NEW: Token expiration fields + token_expires_at: Optional[float] = None + token_expires_in_seconds: Optional[int] = None + refresh_token_expires_at: Optional[float] = None + refresh_token_expires_in_seconds: Optional[int] = None def to_dict(self) -> Dict[str, Any]: - """Convert to API response format""" + """Convert to dictionary representation""" + import time + result = { - # Identification - 'provider': f"{self.provider_name}_{self.country}", - 'provider_name': self.provider_name, - 'provider_label': self.provider_label, - 'country': self.country, - - # Authentication - 'auth_type': self.auth_type, - 'auth_state': self.auth_state.value, - - # Readiness - 'is_ready': self.is_ready, - 'readiness_reason': self.readiness_reason, - - # Tokens - 'primary_token_scope': self.primary_token_scope, - 'has_valid_token': self.has_valid_token, - 'token_scopes': { - scope: token.to_dict() - for scope, token in self.token_scopes.items() + "provider": f"{self.provider_name}_{self.country}" if self.country else self.provider_name, + "provider_name": self.provider_name, + "provider_label": self.provider_label, + "country": self.country, + "auth_type": self.auth_type, + "auth_state": self.auth_state.value, + "is_ready": self.is_ready, + "readiness_reason": self.readiness_reason, + "primary_token_scope": self.primary_token_scope, + "has_valid_token": self.has_valid_token, + "token_scopes": { + scope: { + "scope": info.scope, + "has_token": info.has_token, + "is_valid": info.is_valid, + "expires_at": info.expires_at, + "has_refresh_token": info.has_refresh_token, + "auth_level": info.auth_level + } + for scope, info in self.token_scopes.items() }, - - # Credentials - 'requires_stored_credentials': self.requires_stored_credentials, - 'has_credentials': self.has_credentials, - 'credentials_type': self.credentials_type, - - # Metadata - 'timestamp': self.timestamp, - 'last_authentication': self.last_authentication, - 'provider_specific': self.provider_specific + "requires_stored_credentials": self.requires_stored_credentials, + "has_credentials": self.has_credentials, + "credentials_type": self.credentials_type, + "timestamp": time.time(), + "last_authentication": self.last_authentication, + "provider_specific": self.provider_specific, + "auth_state_description": self._get_auth_state_description() } - # Add state description - descriptions = { - AuthState.NOT_AUTHENTICATED: "Not authenticated", - AuthState.AUTHENTICATED: "Successfully authenticated", - AuthState.EXPIRED: "Authentication expired", - AuthState.PENDING: "Authentication in progress", - AuthState.ERROR: "Authentication error", - AuthState.NOT_APPLICABLE: "Authentication not required" - } - result['auth_state_description'] = descriptions.get(self.auth_state, "Unknown state") + # Add token expiration info if available + if self.token_expires_at is not None: + result["token_expires_at"] = self.token_expires_at - return result \ No newline at end of file + if self.token_expires_in_seconds is not None: + result["token_expires_in_seconds"] = self.token_expires_in_seconds + + if self.refresh_token_expires_at is not None: + result["refresh_token_expires_at"] = self.refresh_token_expires_at + + if self.refresh_token_expires_in_seconds is not None: + result["refresh_token_expires_in_seconds"] = self.refresh_token_expires_in_seconds + + return result + + def _get_auth_state_description(self) -> str: + """Get human-readable description of auth state""" + if self.auth_state == AuthState.NOT_APPLICABLE: + return "Authentication not required" + elif self.auth_state == AuthState.AUTHENTICATED: + return "Successfully authenticated" + elif self.auth_state == AuthState.EXPIRED: + return "Authentication expired, refresh available" + else: + return "Not authenticated" \ No newline at end of file diff --git a/lib/streaming_providers/providers/auth_builder.py b/lib/streaming_providers/providers/auth_builder.py index 708504c..1dc34ee 100644 --- a/lib/streaming_providers/providers/auth_builder.py +++ b/lib/streaming_providers/providers/auth_builder.py @@ -41,6 +41,11 @@ class AuthStatusBuilder: provider, context, current_auth_type ) + # Get token expiration info + (token_expires_at, token_expires_in_seconds, + refresh_token_expires_at, refresh_token_expires_in_seconds) = \ + AuthStatusBuilder._get_token_expiration_info(provider, context) + # Build and return return AuthStatus( provider_name=provider.provider_name, @@ -57,7 +62,11 @@ class AuthStatusBuilder: primary_token_scope=provider.primary_token_scope, token_scopes=token_scopes, last_authentication=AuthStatusBuilder._get_last_auth_time(provider, context), - provider_specific=provider_specific + provider_specific=provider_specific, + token_expires_at=token_expires_at, + token_expires_in_seconds=token_expires_in_seconds, + refresh_token_expires_at=refresh_token_expires_at, + refresh_token_expires_in_seconds=refresh_token_expires_in_seconds ) # ===== Calculation Methods (with provider override support) ===== @@ -268,4 +277,92 @@ class AuthStatusBuilder: custom_details = provider.get_auth_details(context) details.update(custom_details) - return details \ No newline at end of file + return details + + @staticmethod + def _get_token_expiration_info(provider, context: AuthContext) -> Tuple[ + Optional[float], Optional[int], Optional[float], Optional[int] + ]: + """ + Get token expiration information. + + Returns: + Tuple of (token_expires_at, token_expires_in_seconds, + refresh_token_expires_at, refresh_token_expires_in_seconds) + """ + import time + current_time = time.time() + + token_expires_at: Optional[float] = None + token_expires_in_seconds: Optional[int] = None + refresh_token_expires_at: Optional[float] = None + refresh_token_expires_in_seconds: Optional[int] = None + + # Try to get the primary token + token_data: Optional[Dict[str, Any]] = None + + # Check primary scope first + if provider.primary_token_scope: + token_data = context.get_token( + provider.provider_name, + provider.primary_token_scope, + provider.country + ) + + # If no primary scope or no token, check root-level token + if not token_data: + token_data = context.get_token( + provider.provider_name, + None, + provider.country + ) + + # If still no token, try first available scope + if not token_data and provider.token_scopes: + for scope in provider.token_scopes: + token_data = context.get_token( + provider.provider_name, + scope, + provider.country + ) + if token_data: + break + + if not token_data: + return None, None, None, None + + # Calculate access token expiration + # Standard format: expires_in + issued_at + if 'expires_in' in token_data and 'issued_at' in token_data: + expires_in = token_data['expires_in'] + issued_at = token_data['issued_at'] + if isinstance(expires_in, (int, float)) and isinstance(issued_at, (int, float)): + token_expires_at = float(issued_at) + float(expires_in) + token_expires_in_seconds = int(token_expires_at - current_time) + + # yo_digital format: separate access token expiration + elif 'access_token_expires_in' in token_data and 'access_token_issued_at' in token_data: + expires_in = token_data['access_token_expires_in'] + issued_at = token_data['access_token_issued_at'] + if isinstance(expires_in, (int, float)) and isinstance(issued_at, (int, float)): + token_expires_at = float(issued_at) + float(expires_in) + token_expires_in_seconds = int(token_expires_at - current_time) + + # Direct expiration timestamp + elif 'expires_at' in token_data: + expires_at = token_data['expires_at'] + if isinstance(expires_at, (int, float)): + token_expires_at = float(expires_at) + token_expires_in_seconds = int(token_expires_at - current_time) + + # Calculate refresh token expiration (yo_digital format) + if ('refresh_token_expires_in' in token_data and + 'refresh_token_issued_at' in token_data): + refresh_expires_in = token_data['refresh_token_expires_in'] + refresh_issued_at = token_data['refresh_token_issued_at'] + if isinstance(refresh_expires_in, (int, float)) and isinstance(refresh_issued_at, (int, float)): + refresh_token_expires_at = float(refresh_issued_at) + float(refresh_expires_in) + refresh_token_expires_in_seconds = int(refresh_token_expires_at - current_time) + + return (token_expires_at, token_expires_in_seconds, + refresh_token_expires_at, refresh_token_expires_in_seconds) \ No newline at end of file