From c56e69e2d28ab7581d560209a5287615735cbe01 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Thu, 18 Jun 2026 21:09:18 +0200 Subject: [PATCH] restructure epg details --- .../base/epg_operations.py | 11 ++-- .../base/models/epg_models.py | 54 +++++++++++++++++++ lib/streaming_providers/base/provider.py | 4 +- .../providers/magentaeu/epg_manager.py | 42 ++++++++++++--- .../providers/magentaeu/provider.py | 11 ++-- routes/epg.py | 9 +--- 6 files changed, 102 insertions(+), 29 deletions(-) diff --git a/lib/streaming_providers/base/epg_operations.py b/lib/streaming_providers/base/epg_operations.py index 1c576ce..cf5ad73 100644 --- a/lib/streaming_providers/base/epg_operations.py +++ b/lib/streaming_providers/base/epg_operations.py @@ -8,7 +8,7 @@ from typing import Dict, List, Optional from .epg import EPGManager from .utils.logger import logger -from .models.epg_models import EPGEntry +from .models.epg_models import EPGEntry, EPGProgramDetails class EPGOperations: @@ -154,7 +154,7 @@ class EPGOperations: self, provider_name: str, program_id: str, - ) -> Optional["EPGEntry"]: + ) -> Optional[EPGProgramDetails]: """Get full metadata for a single program. Args: @@ -168,10 +168,9 @@ class EPGOperations: 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) + # The generic EPGManager has no program-detail index; native providers + # must implement get_program_details() to support this endpoint. + return None # ------------------------------------------------------------------ # XMLTV export diff --git a/lib/streaming_providers/base/models/epg_models.py b/lib/streaming_providers/base/models/epg_models.py index 15349f9..605666f 100644 --- a/lib/streaming_providers/base/models/epg_models.py +++ b/lib/streaming_providers/base/models/epg_models.py @@ -631,6 +631,59 @@ class EPGEntry: ) +@dataclass(frozen=True) +class EPGProgramDetails: + """ + Enrichment metadata for a single programme fetched from a provider's + program-detail endpoint. + + This is intentionally separate from EPGEntry: it carries only the fields + that a detail endpoint can return on its own — no scheduling context + (broadcast_id, title, start, end) is included because those fields live + in the schedule item, not the detail response. + + Callers that already hold an EPGEntry (e.g. a route handler that looked up + a programme from the EPG grid) can merge these details into their response + without constructing a second full EPGEntry. + """ + + program_id: str + """Provider-scoped programme identifier.""" + + description: Optional[str] = None + """Full programme description/plot.""" + + episode_name: Optional[str] = None + """Episode title/sub-title.""" + + year: Optional[int] = None + """Production/release year.""" + + icon: Optional[str] = None + """Poster or thumbnail URL.""" + + cast: Optional[List[str]] = None + directors: Optional[List[str]] = None + writers: Optional[List[str]] = None + producers: Optional[List[str]] = None + presenter: Optional[List[str]] = None + composers: Optional[List[str]] = None + contributors: Optional[List[str]] = None + + def to_dict(self) -> dict: + """Serialise to a plain dict, omitting None values.""" + result: dict = {"program_id": self.program_id} + for field in ( + "description", "episode_name", "year", "icon", + "cast", "directors", "writers", "producers", + "presenter", "composers", "contributors", + ): + value = getattr(self, field) + if value is not None: + result[field] = value + return result + + # Constants matching C++ EPG_TAG_FLAG values class EPGFlags: """ @@ -893,6 +946,7 @@ PVREPGTag = EPGEntry __all__ = [ # Main classes "EPGEntry", + "EPGProgramDetails", "PVREPGTag", # Legacy alias # Constants "EPG_TAG_INVALID_UID", diff --git a/lib/streaming_providers/base/provider.py b/lib/streaming_providers/base/provider.py index 8a2ddce..639afbb 100644 --- a/lib/streaming_providers/base/provider.py +++ b/lib/streaming_providers/base/provider.py @@ -23,7 +23,7 @@ from .models.subscription import SubscriptionPackage, UserSubscription from .models.recording import Recording from .models.timer import Timer from .models.timer_type import TimerType -from .models.epg_models import EPGEntry +from .models.epg_models import EPGEntry, EPGProgramDetails from .network import HTTPManager, HTTPManagerFactory from .utils.logger import logger @@ -718,7 +718,7 @@ class StreamingProvider(ABC): """ return {} - def get_program_details(self, program_id: str, **kwargs) -> Optional["EPGEntry"]: + def get_program_details(self, program_id: str, **kwargs) -> Optional["EPGProgramDetails"]: """ Get detailed metadata for a single program. diff --git a/lib/streaming_providers/providers/magentaeu/epg_manager.py b/lib/streaming_providers/providers/magentaeu/epg_manager.py index c89f310..99e5a96 100644 --- a/lib/streaming_providers/providers/magentaeu/epg_manager.py +++ b/lib/streaming_providers/providers/magentaeu/epg_manager.py @@ -34,7 +34,7 @@ from zoneinfo import ZoneInfo from typing import Any, Dict, List, Optional, Set, Tuple from ...base.utils.logger import logger -from ...base.models.epg_models import EPGEntry +from ...base.models.epg_models import EPGEntry, EPGProgramDetails from .constants import ( DEFAULT_REQUEST_TIMEOUT, SUPPORTED_COUNTRIES, @@ -323,19 +323,49 @@ class MagentaEUEpgManager: return result - def get_program_details(self, program_id: str) -> Dict[str, Any]: + def get_program_details(self, program_id: str) -> Optional[EPGProgramDetails]: """ Fetch detailed metadata for a single programme. - Returns the raw details dict from the bifrost API (with in-memory - caching), or an empty dict if the programme is not found or the - request fails. + Fetches the raw bifrost detail response (with in-memory caching) and + maps it to an EPGProgramDetails instance. Returns None if the + programme is not found or the request fails. Parameters ---------- program_id: Programme identifier as returned by the schedule API. """ - return self._fetch_program_details(program_id) + raw = self._fetch_program_details(program_id) + if not raw: + return None + + credit_map = self._parse_credits(raw) + + # Description lives under raw["details"]["description"] + details_block = (raw.get("details") or {}) + description = details_block.get("description") or None + + # Release year + year_raw = raw.get("release_year") + try: + year = int(year_raw) if year_raw else None + except (ValueError, TypeError): + year = None + + return EPGProgramDetails( + program_id=program_id, + description=description, + episode_name=raw.get("episode_name") or None, + year=year, + icon=raw.get("poster_image_url") or None, + cast=credit_map.get("cast"), + directors=credit_map.get("directors"), + writers=credit_map.get("writers"), + producers=credit_map.get("producers"), + presenter=credit_map.get("presenter"), + composers=credit_map.get("composers"), + contributors=credit_map.get("contributors"), + ) # ------------------------------------------------------------------ # HTTP helpers diff --git a/lib/streaming_providers/providers/magentaeu/provider.py b/lib/streaming_providers/providers/magentaeu/provider.py index 04c0240..f2ae5c9 100644 --- a/lib/streaming_providers/providers/magentaeu/provider.py +++ b/lib/streaming_providers/providers/magentaeu/provider.py @@ -6,7 +6,7 @@ from typing import ClassVar, Dict, List, Optional, Tuple from ...base.auth import UserPasswordCredentials from ...base.models import DRMConfig, StreamingChannel, Event -from ...base.models.epg_models import EPGEntry +from ...base.models.epg_models import EPGEntry, EPGProgramDetails from ...base.models.proxy_models import ProxyConfig from ...base.network import ProxyConfigManager from ...base.provider import StreamingProvider @@ -347,9 +347,9 @@ class MagentaEUProvider(StreamingProvider): end_time=end_time, ) - def get_program_details(self, program_id: str, **kwargs) -> Optional[EPGEntry]: + def get_program_details(self, program_id: str, **kwargs) -> Optional["EPGProgramDetails"]: """ - Get detailed metadata for a single programme as an EPGEntry object. + Get detailed metadata for a single programme as an EPGProgramDetails object. Returns None if the programme is not found or the request fails. @@ -357,10 +357,7 @@ class MagentaEUProvider(StreamingProvider): ---------- program_id: Programme identifier as returned by the schedule API. """ - details = self.epg_manager.get_program_details(program_id) - if not details: - return None - return EPGEntry.from_dict(details) + return self.epg_manager.get_program_details(program_id) def enrich_channel_data( self, channel: StreamingChannel, **kwargs diff --git a/routes/epg.py b/routes/epg.py index c697872..6b15d99 100644 --- a/routes/epg.py +++ b/routes/epg.py @@ -375,18 +375,11 @@ def setup_epg_routes(app, manager, service): response.content_type = "application/json; charset=utf-8" return {"error": f"Program '{program_id}' not found for provider '{provider}'"} - # Serialize EPGEntry objects to dicts here, at the response - # boundary — epg_ops returns EPGEntry objects (not JSON-safe) - # on the native path. The generic fallback path can return a - # plain dict, so only convert if it's actually an EPGEntry. - if hasattr(program_data, "to_dict"): - program_data = program_data.to_dict() - response.content_type = "application/json; charset=utf-8" return { "provider": provider, "program_id": program_id, - "details": program_data, + "details": program_data.to_dict(), } except ValueError as e: