script.service.ultimate/lib/streaming_providers/base/models/auth.py

# streaming_providers/base/models/auth.py
from dataclasses import dataclass, field
from typing import Dict, Any, Optional
from enum import Enum
import time


class AuthState(Enum):
    """Standardized authentication states"""
    NOT_AUTHENTICATED = "not_authenticated"
    AUTHENTICATED = "authenticated"
    EXPIRED = "expired"
    PENDING = "pending"
    ERROR = "error"
    NOT_APPLICABLE = "not_applicable"


@dataclass
class TokenInfo:
    """Standardized token information"""
    scope: str
    has_token: bool
    is_valid: bool
    expires_at: Optional[float] = None
    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
    auth_type: str
    auth_state: AuthState

    # Readiness
    is_ready: bool
    readiness_reason: Optional[str] = None

    # 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)

    def to_dict(self) -> Dict[str, Any]:
        """Convert to API response format"""
        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()
            },

            # 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
        }

        # 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")

        return result