nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-17 22:52:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

restructure epg

Browse Source
This commit is contained in:
Nirvana
2026-06-15 22:28:42 +02:00
parent 7e9eb772ee
commit 035cc838da
2 changed files with 202 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/base/epg/epg_manager.py
+55 -2
View File
@@ -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,
lib/streaming_providers/base/epg_operations.py
+147 -15
View File
@@ -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)