nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-07 16:32:19 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add discovery

Browse Source
This commit is contained in:
Nirvana
2026-03-16 19:27:30 +01:00
parent ca8ebb16e4
commit 21c4833a09
3 changed files with 64 additions and 196 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magenta2/constants.py
+29 -2
View File
@@ -155,6 +155,9 @@ MAGENTA2_CLIENT_IDS = {
"ios": "21218403-52ec-4a65-abf4-f36a0eadd631",
}
# Client ID used for SMIL/selector manifest requests
SMIL_CLIENT_ID = "a8198f31-b406-4177-8dee-f6216c356c75"
# OAuth2 scopes
MAGENTA2_OAUTH_SCOPE = "openid profile offline_access tvhubs"
@@ -228,9 +231,10 @@ DEFAULT_MAX_RETRIES = 3
DEFAULT_EPG_WINDOW_HOURS = 3
# Cache durations (seconds)
BOOTSTRAP_CACHE_DURATION = 3600 # 1 hour
BOOTSTRAP_CACHE_DURATION = 3600 # 1 hour
OPENID_CONFIG_CACHE_DURATION = 86400 # 24 hours
MANIFEST_CACHE_DURATION = 7200 # 2 hours
MANIFEST_CACHE_DURATION = 7200 # 2 hours
SMIL_CACHE_DURATION = 3600 # 1 hour
# ============================================================================
# Error Codes
@@ -267,6 +271,29 @@ BOOTSTRAP_PARAMS = {"$redirect": "false"}
# REMOVED: Old BOOTSTRAP_KEYS - now handled in config_models.py
# ============================================================================
# Quality Configuration
# ============================================================================
# Ordered quality preference fallback chains used by both live-channel selection
# and VOD playback. Key = desired quality; value = ordered list to try.
QUALITY_FALLBACK: dict = {
"UHDHDR": ["UHDHDR", "UHD", "HD", "SD"],
"UHD": ["UHD", "HD", "SD"],
"HD": ["HD", "SD"],
"SD": ["SD"],
}
# Integer rank per quality label — used for fast channel deduplication when
# multiple entries share the same display number (keep the highest-ranked one).
QUALITY_RANK: dict = {
"SD": 1,
"HD": 2,
"UHD": 3,
"UHDHDR": 4,
"4K": 3, # treat 4K as equivalent to UHD for ranking purposes
}
# ============================================================================
# VOD Configuration
# ============================================================================
lib/streaming_providers/providers/magenta2/provider.py
+29 -179
View File
@@ -4,9 +4,9 @@ import base64
import json
import re
import time
import datetime
import uuid
from datetime import datetime, timedelta
from urllib.parse import quote
from typing import Any, ClassVar, Dict, List, Optional, Tuple
from ...base.models import DRMConfig, DRMSystem, LicenseConfig, StreamingChannel, Event
@@ -20,7 +20,6 @@ from .auth import Magenta2Authenticator, Magenta2Credentials, Magenta2UserCreden
from .config_models import ProviderConfig
from .constants import (
CONTENT_TYPE_LIVE,
CONTENT_TYPE_VOD,
DEFAULT_COUNTRY,
DEFAULT_EPG_WINDOW_HOURS,
DEFAULT_MAX_RETRIES,
@@ -30,11 +29,17 @@ from .constants import (
DRM_SYSTEM_WIDEVINE,
ERROR_CODES,
MAGENTA2_CLIENT_IDS,
MAGENTA2_FALLBACK_ACCOUNT_URI,
MAGENTA2_LOGO,
MAGENTA2_PLATFORMS,
MODE_LIVE,
MODE_VOD,
QUALITY_RANK,
SMIL_CACHE_DURATION,
SMIL_CLIENT_ID,
SUPPORTED_COUNTRIES,
VOD_PREFIX_EPISODE,
VOD_PREFIX_MOVIE_MV,
VOD_PREFIX_MOVIE_SH,
)
from .discovery import DiscoveryService
from .endpoint_manager import EndpointManager
@@ -78,7 +83,7 @@ class Magenta2Provider(StreamingProvider):
# Generate session ID, device ID, and serial number.
# serial_number is stable for the lifetime of this provider instance.
self.session_id = self._generate_uuid()
self.device_id = self._generate_device_id()
self.device_id = self._generate_uuid()
self.serial_number = self._generate_uuid()
# Setup proxy configuration
@@ -288,26 +293,18 @@ class Magenta2Provider(StreamingProvider):
self.device_token = None
self._persona_cache: Optional[PersonaResult] = None
self._smil_cache: Dict[str, Tuple[float, Dict]] = {} # channel_id -> (timestamp, smil_data)
self._smil_cache_ttl = 3600
logger.info("Magenta2 provider initialization completed successfully")
self._smil_cache_ttl = SMIL_CACHE_DURATION
logger.info("Magenta2 provider initialization completed successfully")
@staticmethod
def _generate_uuid() -> str:
"""Generate UUID for session"""
"""Generate a UUID string. Used for session IDs, device IDs, and call IDs."""
return str(uuid.uuid4())
@staticmethod
def _generate_device_id() -> str:
"""Generate device ID"""
return str(uuid.uuid4())
@staticmethod
def _generate_call_id() -> str:
"""Generate call ID for requests"""
return str(uuid.uuid4())
def _generate_call_id(self) -> str:
"""Generate a fresh call ID for each individual request."""
return self._generate_uuid()
def _load_proxy_from_manager(self, config_dir: Optional[str]) -> Optional[ProxyConfig]:
"""Load proxy configuration from ProxyConfigManager"""
@@ -461,11 +458,11 @@ class Magenta2Provider(StreamingProvider):
@property
def supported_auth_types(self) -> List[str]:
return ["network_based"] # ← Change 1
return ["network_based"]
@property
def primary_token_scope(self) -> Optional[str]:
return "persona" # ← Keep this
return "persona"
@property
def token_scopes(self) -> List[str]:
@@ -662,7 +659,7 @@ class Magenta2Provider(StreamingProvider):
}
# Build query string
query_string = "&".join([f"{k}={self._url_encode(v)}" for k, v in params.items()])
query_string = "&".join([f"{k}={quote(v, safe='')}" for k, v in params.items()])
return f"{base_url}/iss?{query_string}"
@@ -674,7 +671,6 @@ class Magenta2Provider(StreamingProvider):
if "entries" not in response_data:
return []
quality_rank = {"SD": 1, "HD": 2, "UHD": 3, "4K": 3}
best_entries = {}
channels = []
@@ -694,9 +690,9 @@ class Magenta2Provider(StreamingProvider):
channels.append(channel)
continue
# Extract quality for comparison
# Extract quality for comparison using the shared QUALITY_RANK constant
quality = station_info.get("dt$quality", "SD")
current_rank = quality_rank.get(quality, 1)
current_rank = QUALITY_RANK.get(quality, 1)
# Check if we need to replace existing entry
existing = best_entries.get(display_number)
@@ -802,7 +798,6 @@ class Magenta2Provider(StreamingProvider):
response.raise_for_status()
channels_data = response.json()
# ✅ USE OPTIMIZED SINGLE-PASS PROCESSING
channels = self._process_channel_stations_response_optimized(
channels_data, prefer_highest_quality
)
@@ -824,130 +819,6 @@ class Magenta2Provider(StreamingProvider):
) -> List[Event]:
return []
def _get_channels_from_mpx_feeds(self) -> List[StreamingChannel]:
"""Get channels from MPX feeds discovered in manifest"""
try:
if not self.endpoint_manager:
return []
entitled_channels_url = self.endpoint_manager.get_endpoint(
"mpx_feed_entitledChannelsFeed"
)
if not entitled_channels_url:
return []
self._ensure_authenticated()
headers = self._get_api_headers(require_auth=True)
response = self.http_manager.get(
entitled_channels_url,
operation="mpx_feed",
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
feed_data = response.json()
return self._parse_mpx_feed_response(feed_data)
except Exception as e:
logger.warning(f"Failed to get channels from MPX feed: {e}")
return []
def _parse_mpx_feed_response(self, feed_data: Dict) -> List[StreamingChannel]:
"""Parse MPX feed response into channels"""
channels = []
entries = feed_data.get("entries", [])
for entry in entries:
try:
title = entry.get("title", "Unknown")
channel_id = entry.get("guid", "").split("/")[-1] if entry.get("guid") else ""
if not channel_id:
continue
channel = Magenta2Channel(
name=title,
channel_id=channel_id,
content_type=CONTENT_TYPE_LIVE,
country=self.country,
raw_data=entry,
)
streaming_channel = channel.to_streaming_channel(provider_name=self.provider_name)
channels.append(streaming_channel)
except Exception as e:
logger.warning(f"Error parsing MPX channel entry: {e}")
return channels
def _process_channels_response(self, response_data: Dict) -> List[StreamingChannel]:
"""Process API response and convert to StreamingChannel objects"""
channels = []
if isinstance(response_data, list):
channel_list = response_data
elif isinstance(response_data, dict):
if "channels" in response_data:
channel_list = response_data["channels"]
elif "data" in response_data:
channel_list = response_data["data"]
else:
channel_list = [response_data]
else:
logger.warning("Unexpected channel response format")
return channels
for channel_data in channel_list:
try:
channel_id = channel_data.get("id", channel_data.get("channelId", ""))
if not channel_id:
logger.warning("Channel missing ID, skipping")
continue
title = channel_data.get("title", channel_data.get("name", "Unknown Channel"))
stream_type = channel_data.get("type", "LIVE")
quality = channel_data.get("quality", "")
logo_url = None
if "logo" in channel_data:
if isinstance(channel_data["logo"], dict) and "url" in channel_data["logo"]:
logo_url = channel_data["logo"]["url"]
elif isinstance(channel_data["logo"], str):
logo_url = channel_data["logo"]
elif "image" in channel_data:
logo_url = channel_data["image"]
content_type = (
CONTENT_TYPE_LIVE if stream_type.upper() == "LIVE" else CONTENT_TYPE_VOD
)
mode = MODE_LIVE if stream_type.upper() == "LIVE" else MODE_VOD
magenta2_channel = Magenta2Channel(
name=title,
channel_id=channel_id,
logo_url=logo_url,
mode=mode,
content_type=content_type,
country=self.country,
raw_data=channel_data,
)
if quality:
magenta2_channel.name = f"{title} ({quality})"
streaming_channel = magenta2_channel.to_streaming_channel(
provider_name=self.provider_name
)
channels.append(streaming_channel)
except Exception as e:
logger.warning(f"Error processing channel data: {e}")
return channels
def get_entitlement_token(self, content_id: str, content_type: str = CONTENT_TYPE_LIVE) -> str:
"""
Get entitlement token using persona_token Basic auth
@@ -1310,7 +1181,6 @@ class Magenta2Provider(StreamingProvider):
# If we get here, no MPD URL was found
logger.warning(f"No MPD URL found in SMIL response for channel {channel_id}")
# Log the full SMIL content for debugging in case of unexpected format
if len(smil_content) < 1000: # Only log if it's reasonably short
logger.debug(f"Full SMIL content: {smil_content}")
else:
@@ -1419,7 +1289,6 @@ class Magenta2Provider(StreamingProvider):
# If the caller passed a Gracenote content id (GN_EP*, GN_MV*, GN_SH*)
# rather than a real MPX mediaId, resolve it first. The SMIL/selector
# endpoint only accepts MPX mediaIds.
from .constants import VOD_PREFIX_EPISODE, VOD_PREFIX_MOVIE_MV, VOD_PREFIX_MOVIE_SH
if (
channel_id.startswith(VOD_PREFIX_EPISODE)
or channel_id.startswith(VOD_PREFIX_MOVIE_MV)
@@ -1451,7 +1320,6 @@ class Magenta2Provider(StreamingProvider):
logger.error(f"No SMIL content received for channel {channel_id}")
return None
# 🚨 ADD VALIDATION: Check if we got actual SMIL content
if len(smil_content.strip()) == 0:
logger.error(f"Empty SMIL content for channel {channel_id}")
return None
@@ -1508,13 +1376,14 @@ class Magenta2Provider(StreamingProvider):
logger.error(f"No account pid!:")
return None
# Use the same client_id as in the original SMIL request
client_id = "a8198f31-b406-4177-8dee-f6216c356c75"
smil_url = f"{selector_service}{account_pid}/media/{channel_id}?format=smil&formats=MPEG-DASH&tracking=true&clientId={client_id}"
smil_url = (
f"{selector_service}{account_pid}/media/{channel_id}"
f"?format=smil&formats=MPEG-DASH&tracking=true&clientId={SMIL_CLIENT_ID}"
)
headers = {
"Authorization": f"Basic {persona_token}",
"User-Agent": self.platform_config["user_agent"], # Platform user agent
"User-Agent": self.platform_config["user_agent"],
"Accept": "application/smil+xml, application/xml;q=0.9, */*;q=0.8",
}
@@ -1522,14 +1391,12 @@ class Magenta2Provider(StreamingProvider):
logger.debug(f"SMIL Request Headers:")
for key, value in headers.items():
if key == "Authorization":
# Don't log full auth token for security, but show it exists
logger.debug(f" {key}: Basic [REDACTED] (length: {len(persona_token)})")
else:
logger.debug(f" {key}: {value}")
# Also debug the persona token structure
try:
# Try to decode to see what's in there
decoded = base64.b64decode(persona_token).decode("utf-8")
logger.debug(f"Decoded persona token preview: {decoded[:100]}...")
except Exception as decode_error:
@@ -1555,8 +1422,8 @@ class Magenta2Provider(StreamingProvider):
extract_and_release_lock(
smil_content,
self.http_manager,
client_id=client_id, # Use the same client_id as SMIL request
user_agent=self.platform_config["user_agent"], # Platform user agent
client_id=SMIL_CLIENT_ID,
user_agent=self.platform_config["user_agent"],
)
return smil_content
@@ -1568,13 +1435,6 @@ class Magenta2Provider(StreamingProvider):
logger.error(f"Error getting SMIL content for DRM: {e}")
return None
@staticmethod
def _url_encode(value: str) -> str:
"""URL encode a string"""
from urllib.parse import quote
return quote(value, safe="")
@staticmethod
def _extract_release_pid_from_smil(smil_content: str) -> Optional[str]:
"""Extract releasePid from SMIL trackingData parameter"""
@@ -1622,7 +1482,7 @@ class Magenta2Provider(StreamingProvider):
logger.error("No colon found in decoded persona token")
return None
persona_jwt = decoded[last_colon_index + 1 :]
persona_jwt = decoded[last_colon_index + 1:]
if not persona_jwt.startswith("eyJ"):
logger.error(f"Extracted token doesn't look like a JWT")
@@ -1645,7 +1505,6 @@ class Magenta2Provider(StreamingProvider):
if not smil_data.get("release_pid"):
logger.error(f"No releasePid found in SMIL for channel {content_id}")
# Debug what we do have
logger.debug(f"SMIL data keys: {list(smil_data.keys())}")
if "content" in smil_data and smil_data["content"]:
logger.debug(f"SMIL content preview: {smil_data['content'][:500]}...")
@@ -1670,7 +1529,7 @@ class Magenta2Provider(StreamingProvider):
# Get account URI
account_uri = self._get_account_uri()
encoded_account_uri = self._url_encode(account_uri)
encoded_account_uri = quote(account_uri, safe="")
# Build license URL with the extracted JWT
license_url = (
@@ -1715,8 +1574,7 @@ class Magenta2Provider(StreamingProvider):
if account_uri:
return account_uri
# Fallback
return "http://access.auth.theplatform.com/data/Account/2709353023"
return MAGENTA2_FALLBACK_ACCOUNT_URI
def get_epg(
self,
@@ -1765,10 +1623,6 @@ class Magenta2Provider(StreamingProvider):
Custom auth state calculation for Magenta2.
Checks persona token with its special structure.
"""
import time
from ...base.models.auth import AuthState
# Get persona token data
persona_token = context.get_token(self.provider_name, "persona", self.country)
@@ -1802,8 +1656,6 @@ class Magenta2Provider(StreamingProvider):
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)
@@ -1853,8 +1705,6 @@ class Magenta2Provider(StreamingProvider):
Provide Magenta2-specific auth details.
Shows status of all token scopes.
"""
import time
details = {}
# Check all token scopes
lib/streaming_providers/providers/magenta2/vod_manager.py
+6 -15
View File
@@ -43,12 +43,14 @@ Public interface
"""
import time
import uuid as _uuid_mod
from typing import Dict, List, Optional, Union
from ...base.utils.logger import logger
from ...base.models.vod import VodCategory, VodItem
from .constants import (
QUALITY_FALLBACK,
SUBSCRIBER_TYPES,
TVHUBS_BASE_URL,
VOD_DEFAULT_PAGE_SIZE,
@@ -60,6 +62,7 @@ from .constants import (
VOD_STREAMING_TILE_TITLES,
)
class VodManager:
"""
Manages Magenta2 VOD catalogue traversal.
@@ -77,16 +80,6 @@ class VodManager:
TVHUBS_BASE_URL fallback constant.
"""
# Quality preference order: each entry is the desired quality, followed by
# progressively lower fallbacks. The first quality in the list that exists
# in a playbackUrls response is used.
_QUALITY_FALLBACK: Dict[str, List[str]] = {
"UHDHDR": ["UHDHDR", "UHD", "HD", "SD"],
"UHD": ["UHD", "HD", "SD"],
"HD": ["HD", "SD"],
"SD": ["SD"],
}
def __init__(
self,
http_manager,
@@ -105,11 +98,10 @@ class VodManager:
# Stable serial number for the lifetime of this manager instance.
# Passed in from the provider so it stays consistent across all
# requests. Falls back to a fresh UUID only when not supplied (e.g. tests).
import uuid as _uuid_mod
self._serial_number: str = serial_number or str(_uuid_mod.uuid4())
# Normalise to uppercase; fall back to "HD" for unknown values.
_q = (preferred_quality or "HD").upper()
self._preferred_quality: str = _q if _q in self._QUALITY_FALLBACK else "HD"
self._preferred_quality: str = _q if _q in QUALITY_FALLBACK else "HD"
# Optional callable() -> Dict[str, str] returning auth headers for
# authenticated VOD endpoints (vodproductinformation, VodPlayer).
# When None, requests are sent without auth (browsing/catalogue only).
@@ -1056,7 +1048,7 @@ class VodManager:
"""
Select the best mediaId from a VodPlayer ``content.playbackUrls`` list.
The quality is chosen by walking ``_QUALITY_FALLBACK[self._preferred_quality]``
The quality is chosen by walking ``QUALITY_FALLBACK[self._preferred_quality]``
in order and returning the first ``mediaId`` whose ``quality`` matches.
If no entry in the fallback chain matches, the first entry in the list is
returned as a last resort (mirrors legacy behaviour).
@@ -1078,7 +1070,7 @@ class VodManager:
if entry.get("quality") and entry.get("mediaId")
}
for q in self._QUALITY_FALLBACK.get(self._preferred_quality, ["HD", "SD"]):
for q in QUALITY_FALLBACK.get(self._preferred_quality, ["HD", "SD"]):
if q in quality_map:
logger.debug(
f"{self._provider}: Selected quality '{q}' "
@@ -1223,7 +1215,6 @@ class VodManager:
def _playback_params(self) -> Dict:
"""Query parameters for vodproductinformation / VodPlayer requests."""
import time
return {
"$deviceModel": self._device_model,
"$profile": self._profile_name,