mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-09-23 17:42:19 +02:00
Add web config
This commit is contained in:
Nirvana
@@ -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
|
||||
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"
|
||||
@@ -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
|
||||
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)
|
||||
Reference in New Issue
Block a user