diff --git a/lib/streaming_providers/base/models/streaming_channel.py b/lib/streaming_providers/base/models/streaming_channel.py index 0493223..3d45df6 100644 --- a/lib/streaming_providers/base/models/streaming_channel.py +++ b/lib/streaming_providers/base/models/streaming_channel.py @@ -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, diff --git a/lib/streaming_providers/providers/magenta2/provider.py b/lib/streaming_providers/providers/magenta2/provider.py index da7fc27..e321139 100644 --- a/lib/streaming_providers/providers/magenta2/provider.py +++ b/lib/streaming_providers/providers/magenta2/provider.py @@ -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: