From 035cc838da32c4c81b030f1f51ecbd851954fe70 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Mon, 15 Jun 2026 22:28:42 +0200 Subject: [PATCH] restructure epg --- .../base/epg/epg_manager.py | 57 +++++- .../base/epg_operations.py | 162 ++++++++++++++++-- 2 files changed, 202 insertions(+), 17 deletions(-) diff --git a/lib/streaming_providers/base/epg/epg_manager.py b/lib/streaming_providers/base/epg/epg_manager.py index bb15d34..f6c738e 100644 --- a/lib/streaming_providers/base/epg/epg_manager.py +++ b/lib/streaming_providers/base/epg/epg_manager.py @@ -142,6 +142,8 @@ class EPGManager: channel_id: str, start_time: Optional[Union[datetime, int, float]] = None, end_time: Optional[Union[datetime, int, float]] = None, + limit: Optional[int] = None, + country: Optional[str] = None, ) -> List[Dict]: """ Get EPG data for a specific channel within a time range. @@ -155,6 +157,8 @@ class EPGManager: channel_id: Channel ID within provider start_time: Start of time range (datetime), None for now end_time: End of time range (datetime), None for now+12h + limit: Maximum number of entries to return (None = no limit) + country: Optional country filter (forwarded to mapping lookup) Returns: List of EPG entries as dictionaries (empty on any error) @@ -163,7 +167,9 @@ class EPGManager: try: # Step 1: Map to EPG channel ID - epg_channel_id = self.mapping.get_epg_channel_id(provider_name, channel_id) + epg_channel_id = self.mapping.get_epg_channel_id( + provider_name, channel_id, country=country + ) if not epg_channel_id: logger.warning(f"EPGManager: No EPG mapping found for {provider_name}/{channel_id}") return [] @@ -198,12 +204,16 @@ class EPGManager: provider_name, # Enable provider encoding ) + # Step 5: Apply limit if requested + if limit is not None and len(epg_entries) > limit: + epg_entries = epg_entries[:limit] + logger.info( f"EPGManager: Retrieved {len(epg_entries)} EPG entries for " f"{provider_name}/{channel_id}" ) - # Step 5: Convert EPGEntry objects to dictionaries for external consumers + # Step 6: Convert EPGEntry objects to dictionaries for external consumers return [entry.to_dict() for entry in epg_entries] except Exception as e: @@ -263,6 +273,49 @@ class EPGManager: logger.error(f"EPGManager: Error getting EPG entries: {e}", exc_info=True) return [] + def get_channel_ids(self, provider_name: str) -> List[str]: + """ + Return all channel IDs that have an EPG mapping for *provider_name*. + + Used by EPGOperations.get_provider_epg_grid() as the fallback channel + list when the caller does not supply an explicit channel_ids filter. + + Args: + provider_name: Registered provider identifier. + + Returns: + List of channel IDs (may be empty if no mapping exists). + """ + provider_mapping = self.mapping.get_provider_mapping(provider_name) + if not provider_mapping: + logger.warning(f"EPGManager: No channels mapped for provider '{provider_name}'") + return [] + return list(provider_mapping.keys()) + + def get_program_by_id(self, program_id: str) -> Optional[Dict]: + """ + Look up a single program by its broadcast/program ID. + + NOTE: The generic EPG manager works on per-channel time-window scans + and does not maintain a global program index. This method is + therefore not supported on the generic path and always returns None. + Providers that implement native EPG (``provider.implements_epg``) + should override this via ``provider.get_program_details(program_id)`` + in EPGOperations instead. + + Args: + program_id: Provider-scoped program identifier. + + Returns: + Always None for the generic path. + """ + logger.debug( + f"EPGManager.get_program_by_id: generic path does not support " + f"program ID lookup (id={program_id!r}); " + "implement get_program_details() on the provider for native support." + ) + return None + def get_epg_for_provider( self, provider_name: str, diff --git a/lib/streaming_providers/base/epg_operations.py b/lib/streaming_providers/base/epg_operations.py index ece79db..043e694 100644 --- a/lib/streaming_providers/base/epg_operations.py +++ b/lib/streaming_providers/base/epg_operations.py @@ -24,30 +24,66 @@ class EPGOperations: self.epg_manager = EPGManager() logger.debug("EPGOperations: Initialized") - def get_channel_epg(self, provider_name: str, channel_id: str, **kwargs) -> List[Dict]: - """Get EPG data for a specific channel.""" + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _get_provider(self, provider_name: str): + """Return the provider instance or raise ValueError.""" provider = self.registry.get_provider(provider_name) if not provider: raise ValueError(f"Provider '{provider_name}' not found or disabled") + return provider + + # ------------------------------------------------------------------ + # Channel EPG + # ------------------------------------------------------------------ + + def get_channel_epg( + self, + provider_name: str, + channel_id: str, + start_time=None, + end_time=None, + limit: int = 100, + country: Optional[str] = None, + ) -> List[Dict]: + """Get EPG data for a specific channel. + + Args: + provider_name: Registered provider identifier. + channel_id: Provider-scoped channel identifier. + start_time: Window start as a timezone-aware datetime (optional). + end_time: Window end as a timezone-aware datetime (optional). + limit: Maximum number of programs to return (default 100). + country: Optional country filter forwarded to the provider. + """ + provider = self._get_provider(provider_name) if provider.implements_epg: logger.debug(f"Using native EPG for '{provider_name}'") - epg_data = provider.get_epg(channel_id, **kwargs) + epg_data = provider.get_epg( + channel_id, + start_time=start_time, + end_time=end_time, + limit=limit, + country=country, + ) else: logger.debug(f"Using generic EPG for '{provider_name}'") epg_data = self.epg_manager.get_epg( provider_name=provider_name, channel_id=channel_id, - start_time=kwargs.get("start_time"), - end_time=kwargs.get("end_time"), + start_time=start_time, + end_time=end_time, + limit=limit, + country=country, ) # Guard against a provider returning None instead of an empty list. - # Without this check the len() call below (and any downstream - # iteration) would raise "NoneType has no attribute …". if epg_data is None: logger.warning( - f"provider.get_epg() returned None for channel '{channel_id}' " + f"get_epg() returned None for channel '{channel_id}' " f"on provider '{provider_name}' — treating as empty result" ) epg_data = [] @@ -55,18 +91,114 @@ class EPGOperations: logger.debug(f"Retrieved {len(epg_data)} EPG entries for '{channel_id}'") return epg_data - def get_provider_epg_xmltv(self, provider_name: str, **kwargs) -> Optional[str]: - """Get complete EPG data in XMLTV format.""" - provider = self.registry.get_provider(provider_name) - if not provider: - raise ValueError(f"Provider '{provider_name}' not found or disabled") + # ------------------------------------------------------------------ + # Multi-channel grid + # ------------------------------------------------------------------ + + def get_provider_epg_grid( + self, + provider_name: str, + start_time=None, + end_time=None, + channel_ids: Optional[List[str]] = None, + country: Optional[str] = None, + ) -> Dict[str, List[Dict]]: + """Get a time-windowed EPG grid across multiple channels. + + Returns a dict keyed by channel_id, each value being a list of + program dicts within the requested window. + + Args: + provider_name: Registered provider identifier. + start_time: Window start as a timezone-aware datetime (optional). + end_time: Window end as a timezone-aware datetime (optional). + channel_ids: Subset of channels to include; None means all. + country: Optional country filter forwarded to the provider. + """ + provider = self._get_provider(provider_name) if provider.implements_epg: - return provider.get_epg_xmltv(**kwargs) + logger.debug(f"Using native EPG grid for '{provider_name}'") + return provider.get_epg_grid( + start_time=start_time, + end_time=end_time, + channel_ids=channel_ids, + country=country, + ) or {} + + # Fallback: fan out to the generic EPG manager per channel. + logger.debug(f"Using generic EPG grid fallback for '{provider_name}'") + channels = channel_ids or self.epg_manager.get_channel_ids(provider_name) + grid: Dict[str, List[Dict]] = {} + for cid in channels: + entries = self.epg_manager.get_epg( + provider_name=provider_name, + channel_id=cid, + start_time=start_time, + end_time=end_time, + ) or [] + grid[cid] = entries + + logger.debug( + f"Grid for '{provider_name}': {len(grid)} channels, " + f"window {start_time} – {end_time}" + ) + return grid + + # ------------------------------------------------------------------ + # Program detail + # ------------------------------------------------------------------ + + def get_program_details( + self, + provider_name: str, + program_id: str, + ) -> Optional[Dict]: + """Get full metadata for a single program. + + Args: + provider_name: Registered provider identifier. + program_id: Provider-scoped program identifier. + """ + provider = self._get_provider(provider_name) + + if provider.implements_epg: + logger.debug(f"Using native program detail for '{provider_name}/{program_id}'") + return provider.get_program_details(program_id) + + logger.debug(f"Using generic program detail for '{provider_name}/{program_id}'") + # EPGManager.get_program_by_id() always returns None on the generic + # path — there is no global program index. Native providers should + # implement get_program_details() to support this endpoint. + return self.epg_manager.get_program_by_id(program_id) + + # ------------------------------------------------------------------ + # XMLTV export + # ------------------------------------------------------------------ + + def get_provider_epg_xmltv( + self, + provider_name: str, + country: Optional[str] = None, + ) -> Optional[str]: + """Get the complete EPG feed in XMLTV format. + + Args: + provider_name: Registered provider identifier. + country: Optional country filter forwarded to the provider. + """ + provider = self._get_provider(provider_name) + + if provider.implements_epg: + return provider.get_epg_xmltv(country=country) logger.warning(f"Provider '{provider_name}' has no XMLTV EPG") return None + # ------------------------------------------------------------------ + # Cache / mapping utilities + # ------------------------------------------------------------------ + def clear_epg_cache(self) -> bool: """Clear the generic EPG cache.""" return self.epg_manager.clear_cache() @@ -84,5 +216,5 @@ class EPGOperations: return self.epg_manager.get_mapping_stats() def has_epg_mapping(self, provider_name: str, channel_id: str) -> bool: - """Check if EPG mapping exists.""" + """Check if an EPG mapping exists for the given channel.""" return self.epg_manager.has_mapping_for_channel(provider_name, channel_id) \ No newline at end of file