mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-09-18 23:22:38 +02:00
Fix get_manifest
This commit is contained in:
Nirvana
@@ -16,6 +16,10 @@ class StreamingChannel:
|
||||
# Visual elements
|
||||
logo_url: Optional[str] = None
|
||||
|
||||
# Channel metadata
|
||||
channel_number: Optional[int] = None # Display channel number (e.g., 1, 3, 99)
|
||||
quality: Optional[str] = None # 'SD', 'HD', 'UHD', '4K'
|
||||
|
||||
# Streaming configuration
|
||||
mode: str = "live" # "live" or "vod"
|
||||
session_manifest: bool = False
|
||||
@@ -55,6 +59,8 @@ class StreamingChannel:
|
||||
'Id': self.channel_id,
|
||||
'Provider': self.provider,
|
||||
'LogoUrl': self.logo_url,
|
||||
'ChannelNumber': self.channel_number,
|
||||
'Quality': self.quality,
|
||||
'Mode': self.mode,
|
||||
'SessionManifest': self.session_manifest,
|
||||
'Manifest': self.manifest,
|
||||
|
||||
@@ -494,82 +494,6 @@ class Magenta2Provider(StreamingProvider):
|
||||
def get_dynamic_manifest_params(self, channel: StreamingChannel, **kwargs) -> Optional[str]:
|
||||
return None
|
||||
|
||||
def _process_channel_stations_response(self, response_data: Dict) -> List[StreamingChannel]:
|
||||
"""Process channel stations feed response"""
|
||||
channels = []
|
||||
|
||||
if 'entries' not in response_data:
|
||||
logger.warning("No entries found in channel stations response")
|
||||
return channels
|
||||
|
||||
for entry in response_data['entries']:
|
||||
try:
|
||||
# Extract station information first
|
||||
stations = entry.get('stations', {})
|
||||
station_info = None
|
||||
if stations:
|
||||
# Get the first station (there's usually only one)
|
||||
station_id = next(iter(stations.keys()))
|
||||
station_info = stations[station_id]
|
||||
|
||||
# PREFER station title over entry title (cleaner name)
|
||||
title = None
|
||||
if station_info:
|
||||
title = station_info.get('title') # "RNF HD"
|
||||
|
||||
if not title:
|
||||
title = entry.get('title', 'Unknown Channel') # Fallback: "RNF HD - Main"
|
||||
|
||||
# Remove "- Main" suffix if present
|
||||
if title and " - Main" in title:
|
||||
title = title.replace(" - Main", "")
|
||||
|
||||
# Extract CORRECT channel ID from era$mediaPids
|
||||
channel_id = self._extract_channel_id_from_entry(entry)
|
||||
|
||||
if not channel_id:
|
||||
logger.warning(f"No channel ID found for entry: {title}")
|
||||
continue
|
||||
|
||||
# Extract logo URLs from station info
|
||||
original_logo_url = None
|
||||
if station_info:
|
||||
thumbnails = station_info.get('thumbnails', {})
|
||||
if 'stationLogo' in thumbnails:
|
||||
original_logo_url = thumbnails['stationLogo'].get('url')
|
||||
elif 'stationLogoColored' in thumbnails:
|
||||
original_logo_url = thumbnails['stationLogoColored'].get('url')
|
||||
|
||||
# BUILD SCALED LOGO URL
|
||||
logo_url = None
|
||||
if original_logo_url:
|
||||
logo_url = self._build_scaled_image_url(original_logo_url)
|
||||
logger.debug(f"Logo URL scaled: {original_logo_url} -> {logo_url}")
|
||||
|
||||
# Create channel object with scaled logo URL
|
||||
magenta2_channel = Magenta2Channel(
|
||||
name=title,
|
||||
channel_id=channel_id,
|
||||
logo_url=logo_url, # Use scaled URL
|
||||
mode=MODE_LIVE,
|
||||
content_type=CONTENT_TYPE_LIVE,
|
||||
country=self.country,
|
||||
raw_data=entry
|
||||
)
|
||||
|
||||
streaming_channel = magenta2_channel.to_streaming_channel(
|
||||
provider_name=self.provider_name
|
||||
)
|
||||
channels.append(streaming_channel)
|
||||
|
||||
logger.debug(f"Processed channel: {title} (ID: {channel_id})")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error processing channel station entry: {e}")
|
||||
|
||||
logger.info(f"Successfully processed {len(channels)} channels from channel stations feed")
|
||||
return channels
|
||||
|
||||
@staticmethod
|
||||
def _extract_channel_id_from_entry(entry: Dict) -> Optional[str]:
|
||||
"""Extract the correct channel ID from era$mediaPids"""
|
||||
@@ -641,110 +565,196 @@ class Magenta2Provider(StreamingProvider):
|
||||
|
||||
return f"{base_url}/iss?{query_string}"
|
||||
|
||||
@staticmethod
|
||||
def _filter_channels_by_quality(
|
||||
channels: List[StreamingChannel],
|
||||
prefer_highest_quality: bool = True
|
||||
) -> List[StreamingChannel]:
|
||||
"""
|
||||
Filter duplicate channels by quality preference.
|
||||
def _process_channel_stations_response(self, response_data: Dict, prefer_highest_quality: bool = True) -> List[
|
||||
StreamingChannel]:
|
||||
"""Process channel stations feed response with quality filtering"""
|
||||
channels = []
|
||||
raw_entries = []
|
||||
|
||||
When multiple channels share the same channel number (dt$displayChannelNumber),
|
||||
if 'entries' not in response_data:
|
||||
logger.warning("No entries found in channel stations response")
|
||||
return channels
|
||||
|
||||
# First pass: collect all raw entries with their metadata
|
||||
for entry in response_data['entries']:
|
||||
try:
|
||||
# Extract station information first
|
||||
stations = entry.get('stations', {})
|
||||
if not stations:
|
||||
continue
|
||||
|
||||
station_id = next(iter(stations.keys()))
|
||||
station_info = stations[station_id]
|
||||
|
||||
# Extract display channel number and quality
|
||||
display_number = entry.get('dt$displayChannelNumber')
|
||||
quality = station_info.get('dt$quality', 'SD')
|
||||
|
||||
# PREFER station title over entry title (cleaner name)
|
||||
title = station_info.get('title') if station_info else None
|
||||
if not title:
|
||||
title = entry.get('title', 'Unknown Channel')
|
||||
|
||||
# Remove "- Main" suffix if present
|
||||
if title and " - Main" in title:
|
||||
title = title.replace(" - Main", "")
|
||||
|
||||
# Extract CORRECT channel ID from era$mediaPids
|
||||
channel_id = self._extract_channel_id_from_entry(entry)
|
||||
if not channel_id:
|
||||
logger.warning(f"No channel ID found for entry: {title}")
|
||||
continue
|
||||
|
||||
# Extract logo URLs from station info
|
||||
original_logo_url = None
|
||||
if station_info:
|
||||
thumbnails = station_info.get('thumbnails', {})
|
||||
if 'stationLogo' in thumbnails:
|
||||
original_logo_url = thumbnails['stationLogo'].get('url')
|
||||
elif 'stationLogoColored' in thumbnails:
|
||||
original_logo_url = thumbnails['stationLogoColored'].get('url')
|
||||
|
||||
# BUILD SCALED LOGO URL
|
||||
logo_url = None
|
||||
if original_logo_url:
|
||||
logo_url = self._build_scaled_image_url(original_logo_url)
|
||||
|
||||
# Store raw entry with metadata for filtering
|
||||
raw_entries.append({
|
||||
'entry': entry,
|
||||
'title': title,
|
||||
'channel_id': channel_id,
|
||||
'logo_url': logo_url,
|
||||
'display_number': display_number,
|
||||
'quality': quality,
|
||||
'station_info': station_info
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error processing channel station entry: {e}")
|
||||
|
||||
# Second pass: filter by quality
|
||||
filtered_entries = self._filter_entries_by_quality(raw_entries, prefer_highest_quality)
|
||||
|
||||
# Third pass: convert filtered entries to StreamingChannel objects
|
||||
for entry_data in filtered_entries:
|
||||
try:
|
||||
magenta2_channel = Magenta2Channel(
|
||||
name=entry_data['title'],
|
||||
channel_id=entry_data['channel_id'],
|
||||
logo_url=entry_data['logo_url'],
|
||||
mode=MODE_LIVE,
|
||||
content_type=CONTENT_TYPE_LIVE,
|
||||
country=self.country,
|
||||
raw_data=entry_data['entry']
|
||||
)
|
||||
|
||||
streaming_channel = magenta2_channel.to_streaming_channel(
|
||||
provider_name=self.provider_name
|
||||
)
|
||||
|
||||
# ✅ SET CHANNEL NUMBER AND QUALITY on StreamingChannel
|
||||
streaming_channel.channel_number = entry_data['display_number']
|
||||
streaming_channel.quality = entry_data['quality']
|
||||
|
||||
channels.append(streaming_channel)
|
||||
|
||||
logger.debug(
|
||||
f"Processed channel: {entry_data['title']} "
|
||||
f"(ID: {entry_data['channel_id']}, Number: {entry_data['display_number']}, Quality: {entry_data['quality']})"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"Error converting channel entry: {e}")
|
||||
|
||||
logger.info(f"Successfully processed {len(channels)} channels from channel stations feed")
|
||||
return channels
|
||||
|
||||
@staticmethod
|
||||
def _filter_entries_by_quality(
|
||||
entries: List[Dict],
|
||||
prefer_highest_quality: bool = True
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
Filter duplicate channel entries by quality preference.
|
||||
|
||||
When multiple entries share the same channel number (display_number),
|
||||
keeps only the one with preferred quality.
|
||||
|
||||
Quality hierarchy: UHD > HD > SD
|
||||
|
||||
Args:
|
||||
channels: List of channels to filter
|
||||
entries: List of entry dictionaries with 'display_number' and 'quality' keys
|
||||
prefer_highest_quality: If True, prefer UHD>HD>SD. If False, prefer SD>HD>UHD
|
||||
|
||||
Returns:
|
||||
Filtered list of channels with duplicates removed based on quality preference
|
||||
Filtered list of entry dictionaries
|
||||
"""
|
||||
if not entries:
|
||||
return entries
|
||||
|
||||
# Define quality ranking (higher number = better quality)
|
||||
quality_rank = {
|
||||
'SD': 1,
|
||||
'HD': 2,
|
||||
'UHD': 3
|
||||
'UHD': 3,
|
||||
'4K': 3 # Treat 4K same as UHD
|
||||
}
|
||||
|
||||
# Group channels by display channel number
|
||||
channels_by_number = {}
|
||||
# Group entries by display channel number
|
||||
entries_by_number = {}
|
||||
entries_without_number = []
|
||||
|
||||
for channel in channels:
|
||||
display_number = None
|
||||
if hasattr(channel, 'raw_data') and channel.raw_data:
|
||||
display_number = channel.raw_data.get('dt$displayChannelNumber')
|
||||
logger.debug(
|
||||
f"Channel: {channel.name}, Display Number: {display_number}, Has raw_data: {channel.raw_data is not None}")
|
||||
for entry in entries:
|
||||
display_number = entry.get('display_number')
|
||||
|
||||
if display_number is None:
|
||||
# If no display number, keep the channel (don't filter it)
|
||||
# Use a unique key to ensure it's not grouped with others
|
||||
unique_key = f"_no_number_{id(channel)}"
|
||||
channels_by_number[unique_key] = [channel]
|
||||
entries_without_number.append(entry)
|
||||
continue
|
||||
|
||||
if display_number not in channels_by_number:
|
||||
channels_by_number[display_number] = []
|
||||
channels_by_number[display_number].append(channel)
|
||||
if display_number not in entries_by_number:
|
||||
entries_by_number[display_number] = []
|
||||
|
||||
# Filter each group, keeping only the channel with preferred quality
|
||||
filtered_channels = []
|
||||
# Add quality rank for easier comparison
|
||||
quality = entry.get('quality', 'SD')
|
||||
entry['quality_rank'] = quality_rank.get(quality, quality_rank['SD'])
|
||||
entries_by_number[display_number].append(entry)
|
||||
|
||||
for display_number, channel_group in channels_by_number.items():
|
||||
if len(channel_group) == 1:
|
||||
# Only one channel with this number, keep it
|
||||
filtered_channels.append(channel_group[0])
|
||||
# Filter each group, keeping only the entry with preferred quality
|
||||
filtered_entries = []
|
||||
duplicate_count = 0
|
||||
|
||||
for display_number, entry_group in entries_by_number.items():
|
||||
if len(entry_group) == 1:
|
||||
filtered_entries.append(entry_group[0])
|
||||
continue
|
||||
|
||||
# Multiple channels with same number - select by quality
|
||||
best_channel = None
|
||||
best_quality_rank = -1 if prefer_highest_quality else float('inf')
|
||||
# Multiple entries with same number - select by quality
|
||||
if prefer_highest_quality:
|
||||
best = max(entry_group, key=lambda x: x['quality_rank'])
|
||||
else:
|
||||
best = min(entry_group, key=lambda x: x['quality_rank'])
|
||||
|
||||
for channel in channel_group:
|
||||
# Extract quality from station data in raw_data
|
||||
quality = None
|
||||
if hasattr(channel, 'raw_data') and channel.raw_data:
|
||||
stations = channel.raw_data.get('stations', {})
|
||||
if stations:
|
||||
# Get first station
|
||||
station_id = next(iter(stations.keys()))
|
||||
station_info = stations[station_id]
|
||||
quality = station_info.get('dt$quality')
|
||||
filtered_entries.append(best)
|
||||
duplicate_count += len(entry_group) - 1
|
||||
|
||||
# Get quality rank (default to SD if not found)
|
||||
current_rank = quality_rank.get(quality, quality_rank['SD'])
|
||||
# Log which variant was selected
|
||||
qualities = ', '.join(e['quality'] or 'Unknown' for e in entry_group)
|
||||
logger.debug(
|
||||
f"Channel #{display_number} ('{best['title']}'): "
|
||||
f"Selected {best['quality']} from {len(entry_group)} variants ({qualities})"
|
||||
)
|
||||
|
||||
# Select based on preference
|
||||
if prefer_highest_quality:
|
||||
# Want highest quality (highest rank)
|
||||
if current_rank > best_quality_rank:
|
||||
best_quality_rank = current_rank
|
||||
best_channel = channel
|
||||
else:
|
||||
# Want lowest quality (lowest rank)
|
||||
if current_rank < best_quality_rank:
|
||||
best_quality_rank = current_rank
|
||||
best_channel = channel
|
||||
# Add back entries without display numbers
|
||||
filtered_entries.extend(entries_without_number)
|
||||
|
||||
if best_channel:
|
||||
filtered_channels.append(best_channel)
|
||||
logger.debug(
|
||||
f"Channel number {display_number}: Selected quality "
|
||||
f"{'UHD' if best_quality_rank == 3 else 'HD' if best_quality_rank == 2 else 'SD'} "
|
||||
f"from {len(channel_group)} variants"
|
||||
)
|
||||
if duplicate_count > 0:
|
||||
logger.info(
|
||||
f"Quality filtering: {len(entries)} entries → {len(filtered_entries)} entries "
|
||||
f"(removed {duplicate_count} duplicates, preference: {'highest' if prefer_highest_quality else 'lowest'} quality)"
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Quality filtering: {len(channels)} channels → {len(filtered_channels)} channels "
|
||||
f"(removed {len(channels) - len(filtered_channels)} duplicates, "
|
||||
f"preference: {'highest' if prefer_highest_quality else 'lowest'} quality)"
|
||||
)
|
||||
|
||||
return filtered_channels
|
||||
|
||||
# Update the fetch_channels method signature and implementation:
|
||||
return filtered_entries
|
||||
|
||||
def fetch_channels(self,
|
||||
time_window_hours: int = DEFAULT_EPG_WINDOW_HOURS,
|
||||
@@ -764,23 +774,21 @@ class Magenta2Provider(StreamingProvider):
|
||||
**kwargs: Additional arguments
|
||||
|
||||
Returns:
|
||||
List of StreamingChannel objects, filtered by quality preference
|
||||
List of StreamingChannel objects with channel_number and quality populated,
|
||||
filtered by quality preference
|
||||
"""
|
||||
try:
|
||||
# Get headers WITHOUT requiring auth (lazy auth will happen later if needed)
|
||||
headers = self._get_api_headers(require_auth=False)
|
||||
|
||||
# FIXED: Use the discovered channel stations endpoint
|
||||
# Use the discovered channel stations endpoint
|
||||
url = None
|
||||
if self.endpoint_manager:
|
||||
# Try channel_stations endpoint first (this is the main channel list)
|
||||
url = self.endpoint_manager.get_endpoint('channel_stations')
|
||||
|
||||
# If not found, try other channel endpoints
|
||||
if not url:
|
||||
url = self.endpoint_manager.get_endpoint('channel_list')
|
||||
|
||||
# If still not found, try MPX feeds
|
||||
if not url and self.endpoint_manager.has_endpoint('mpx_feed_entitledChannelsFeed'):
|
||||
url = self.endpoint_manager.get_endpoint('mpx_feed_entitledChannelsFeed')
|
||||
|
||||
@@ -800,13 +808,13 @@ class Magenta2Provider(StreamingProvider):
|
||||
response.raise_for_status()
|
||||
channels_data = response.json()
|
||||
|
||||
# Process the channel stations feed response
|
||||
channels = self._process_channel_stations_response(channels_data)
|
||||
# Process response with quality filtering (populates channel_number and quality)
|
||||
channels = self._process_channel_stations_response(channels_data, prefer_highest_quality)
|
||||
|
||||
# Apply quality-based filtering
|
||||
channels = self._filter_channels_by_quality(channels, prefer_highest_quality)
|
||||
|
||||
logger.info(f"Successfully fetched {len(channels)} channels for country {self.country}")
|
||||
logger.info(
|
||||
f"Successfully fetched {len(channels)} channels for country {self.country} "
|
||||
f"(quality preference: {'highest' if prefer_highest_quality else 'lowest'})"
|
||||
)
|
||||
return channels
|
||||
|
||||
except Exception as e:
|
||||
|
||||
Reference in New Issue
Block a user