diff --git a/lib/streaming_providers/base/utils/mpd_cache.py b/lib/streaming_providers/base/utils/mpd_cache.py index f2b1d92..09dee38 100644 --- a/lib/streaming_providers/base/utils/mpd_cache.py +++ b/lib/streaming_providers/base/utils/mpd_cache.py @@ -31,42 +31,55 @@ class MPDCacheManager: """Get filename for cache metadata""" return f"{cache_key}.meta" - def get(self, provider: str, channel_id: str) -> Optional[str]: + def get(self, provider: str, channel_id: str, max_stale: int = 0) -> Optional[str]: """ - Get cached MPD manifest if valid + Get cached MPD manifest if valid. Args: provider: Provider name channel_id: Channel ID + max_stale: If > 0, still return manifest content up to this many + seconds past expiry instead of None/deleting it. The + caller is expected to only pass this when a live + refetch has already failed — a stale manifest is a + fallback of last resort, not a normal cache hit. + Default 0 preserves existing behaviour everywhere + this is called without the new argument. Returns: - Cached MPD content if valid and not expired, None otherwise + Cached MPD content if valid (or within max_stale), None otherwise """ cache_key = self._get_cache_key(provider, channel_id) meta_file = self._get_meta_filename(cache_key) manifest_file = self._get_manifest_filename(cache_key) try: - # Read metadata first (small file) meta = self.vfs.read_json(meta_file) if not meta: logger.debug(f"No cache metadata found for {cache_key}") return None - # Check expiry expiry = meta.get("expiry", 0) now = int(time.time()) if now >= expiry: + staleness = now - expiry + if max_stale and staleness <= max_stale: + manifest_content = self.vfs.read_text(manifest_file) + if manifest_content: + logger.warning( + f"Serving STALE cache for {cache_key} " + f"({staleness}s past expiry, max_stale={max_stale}s)" + ) + return manifest_content + # meta exists but file missing — nothing to serve stale logger.debug( - f"Cache expired for {cache_key} (expired {now - expiry}s ago)" + f"Cache expired for {cache_key} (expired {staleness}s ago)" ) - # Clean up expired cache self.vfs.delete(manifest_file) self.vfs.delete(meta_file) return None - # Cache is valid, read manifest manifest_content = self.vfs.read_text(manifest_file) if manifest_content: logger.info(f"Cache hit for {cache_key} (expires in {expiry - now}s)") diff --git a/service.py b/service.py index e27efb8..c1b84af 100644 --- a/service.py +++ b/service.py @@ -77,6 +77,8 @@ class UltimateService: self.mpd_cache = MPDCacheManager() logger.info(f"MPD cache initialized: {self.mpd_cache.vfs.base_path}") + self._decrypted_cache = {} # key -> (content, expiry_ts) — memory-only, never persisted + # 1. Determine EPG URL FIRST (with proper precedence) self.epg_url = self._determine_epg_url() logger.info(f"UltimateService: Final EPG URL determined: {self.epg_url}") @@ -393,234 +395,58 @@ class UltimateService: return manifest_response.text, ttl, provider_proxy_url, segment_headers, manifest_response.url - def get_decrypted_manifest( - self, provider: str, channel_id: str, keyids: dict, - highest_quality_only: bool = False, receiver_side: bool = False, - drm_variant: str = "auto", - ) -> str: + def _get_decrypted_cached(self, key: str, max_stale: int = 0) -> Optional[str]: + entry = self._decrypted_cache.get(key) + if not entry: + return None + content, expiry = entry + now = time.time() + if now < expiry: + return content + if max_stale and (now - expiry) <= max_stale: + logger.warning(f"Serving STALE decrypted manifest for {key} ({now - expiry:.1f}s past expiry)") + return content + del self._decrypted_cache[key] + return None + + def _set_decrypted_cached(self, key: str, content: str, ttl: int) -> None: + self._decrypted_cache[key] = (content, time.time() + ttl) + + @staticmethod + def _fetch_and_cache_manifest(fetch_fn, cache_set, cache_get_stale, *, + cacheable: bool, stale_window: int, label: str) -> str: """ - Get rewritten MPD manifest for decrypted playback via media proxy. - Similar to get_proxied_manifest but adds kid/key parameters. + Shared fetch -> rewrite -> cache -> stale-fallback flow. - Args: - provider: Provider name - channel_id: Channel ID - keyids: Dictionary of kid:key pairs - highest_quality_only: If True, keep only highest quality video representation - receiver_side: If True, inject ClearKey signaling for receiver-side decryption - drm_variant: 'auto' (provider default) or 'software' (prefer ClearKey manifest). - Must match the variant used in the original _resolve_stream call so - the second manifest fetch selects the same stream as the first. + fetch_fn() -> (content: str, ttl: int), raises on failure + cache_set(content, ttl) -> None + cache_get_stale(max_stale) -> Optional[str] - Returns: - Rewritten MPD content as string + Caller is responsible for the upfront "is it already cached" check and + manifest_url/config validation before calling this — that keeps a warm + cache hit from paying for a manifest_url lookup it doesn't need. """ - country = request.query.get("country") - - # Note: We don't cache decrypted manifests as they contain keys - logger.info( - f"Generating {'receiver-side clearkey' if receiver_side else 'decrypted'} manifest " - f"for {provider}/{channel_id} (highest_quality_only={highest_quality_only}, " - f"drm_variant={drm_variant})" - ) - - manifest_url = self.manager.get_channel_manifest( - provider_name=provider, channel_id=channel_id, country=country, - drm_variant=drm_variant, - ) - if not manifest_url: - response.status = 404 - response.content_type = "application/json" - return json.dumps( - {"error": f'Manifest not available for channel "{channel_id}" from provider "{provider}"'} - ) - try: - manifest_text, _ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( - provider, channel_id, manifest_url - ) - - rewriter = MPDRewriter( - self.media_proxy_url, - provider_proxy_url, - keyids, - highest_quality_only, - provider=provider, - channel=channel_id, - clearkey_receiver_side=receiver_side, - segment_headers=segment_headers, - ) - rewritten_mpd = rewriter.rewrite_mpd(manifest_text, effective_url) - + content, ttl = fetch_fn() + if cacheable: + cache_set(content, ttl) response.content_type = "application/dash+xml; charset=utf-8" - return rewritten_mpd - + return content except ValueError as e: + # Config error (e.g. no HTTP manager) — not transient, don't mask with stale cache logger.error(str(e)) response.status = 502 response.content_type = "application/json" return json.dumps({"error": str(e)}) except Exception as fetch_err: - logger.error(f"Failed to fetch manifest for decryption: {fetch_err}") - response.status = 502 - response.content_type = "application/json" - return json.dumps({"error": f"Failed to fetch manifest: {str(fetch_err)}"}) - - def get_proxied_catchup_manifest( - self, - provider: str, - channel_id: str, - start_time: int, - end_time: int, - epg_id: str = None, - country: str = None, - ) -> str: - """ - Get proxied and rewritten MPD manifest for catchup content using media proxy. - Similar to get_proxied_manifest but for catchup streams. - """ - cache_key = f"{channel_id}_catchup_{start_time}_{end_time}" - - cached_mpd = self.mpd_cache.get(provider, cache_key) - if cached_mpd: - response.content_type = "application/dash+xml; charset=utf-8" - return cached_mpd - - logger.info(f"Cache miss for catchup {provider}/{channel_id}, fetching manifest") - - if not self.media_proxy_url: - response.status = 503 - response.content_type = "application/json" - return json.dumps({"error": "Media proxy not configured (MEDIA_PROXY_URL not set)"}) - - manifest_url = self.manager.get_catchup_manifest( - provider_name=provider, - channel_id=channel_id, - start_time=start_time, - end_time=end_time, - epg_id=epg_id, - country=country, - ) - if not manifest_url: - response.status = 404 - response.content_type = "application/json" - return json.dumps({"error": "Catchup manifest not available"}) - - try: - manifest_text, ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( - provider, channel_id, manifest_url - ) - - rewriter = MPDRewriter( - self.media_proxy_url, - provider_proxy_url, - None, # No keyids for catchup streams - False, # highest_quality_only — not needed for catchup - provider=provider, - channel=channel_id, - segment_headers=segment_headers, - ) - rewritten_mpd = rewriter.rewrite_mpd(manifest_text, effective_url) - - self.mpd_cache.set( - provider=provider, - channel_id=cache_key, - mpd_content=rewritten_mpd, - ttl=ttl, - original_url=manifest_url, - ) - - response.content_type = "application/dash+xml; charset=utf-8" - return rewritten_mpd - - except ValueError as e: - logger.error(str(e)) - response.status = 502 - response.content_type = "application/json" - return json.dumps({"error": str(e)}) - except Exception as fetch_err: - logger.error(f"Failed to fetch catchup manifest: {fetch_err}") - response.status = 502 - response.content_type = "application/json" - return json.dumps({"error": f"Failed to fetch manifest: {str(fetch_err)}"}) - - def get_decrypted_catchup_manifest( - self, - provider: str, - channel_id: str, - start_time: int, - end_time: int, - keyids: dict, - epg_id: str = None, - highest_quality_only: bool = False, - ) -> str: - """ - Get a server-side-decrypted MPD manifest for catchup/DVR playback. - - Mirrors get_decrypted_manifest() but fetches the catchup manifest URL - (via manager.get_catchup_manifest) rather than the live channel URL. - receiver_side is always False — the /decrypted/ endpoint contract is - that the server performs decryption, never the client. - - Args: - provider: Provider name. - channel_id: Channel ID. - start_time: Catchup window start (Unix timestamp). - end_time: Catchup window end (Unix timestamp). - keyids: dict of {kid: key} ClearKey pairs. - epg_id: Optional EPG programme ID forwarded to the - provider for asset-based catchup resolution. - highest_quality_only: If True, strip all but the highest-quality - video representation from the manifest. - """ - country = request.query.get("country") - - logger.info( - f"Generating decrypted catchup manifest for {provider}/{channel_id} " - f"(start={start_time} end={end_time} highest_quality_only={highest_quality_only})" - ) - - manifest_url = self.manager.get_catchup_manifest( - provider_name=provider, - channel_id=channel_id, - start_time=start_time, - end_time=end_time, - epg_id=epg_id, - country=country, - ) - if not manifest_url: - response.status = 404 - response.content_type = "application/json" - return json.dumps( - {"error": f'Catchup manifest not available for channel "{channel_id}" from provider "{provider}"'} - ) - - try: - manifest_text, _ttl, provider_proxy_url, segment_headers, effective_url = \ - self.fetch_manifest_for_rewriter(provider, channel_id, manifest_url) - - rewriter = MPDRewriter( - self.media_proxy_url, - provider_proxy_url, - keyids, - highest_quality_only, - provider=provider, - channel=channel_id, - clearkey_receiver_side=False, # server decrypts — never inject ClearKey CP - segment_headers=segment_headers, - ) - rewritten_mpd = rewriter.rewrite_mpd(manifest_text, effective_url) - - response.content_type = "application/dash+xml; charset=utf-8" - return rewritten_mpd - - except ValueError as e: - logger.error(str(e)) - response.status = 502 - response.content_type = "application/json" - return json.dumps({"error": str(e)}) - except Exception as fetch_err: - logger.error(f"Failed to fetch catchup manifest for decryption: {fetch_err}") + logger.warning(f"{label} fetch failed: {fetch_err} — checking stale cache") + if cacheable: + stale = cache_get_stale(stale_window) + if stale: + logger.warning(f"Serving stale {label} due to fetch failure") + response.content_type = "application/dash+xml; charset=utf-8" + return stale + logger.error(f"Failed to fetch {label}: {fetch_err}") response.status = 502 response.content_type = "application/json" return json.dumps({"error": f"Failed to fetch manifest: {str(fetch_err)}"}) @@ -648,39 +474,167 @@ class UltimateService: return json.dumps( {"error": f'Manifest not available for channel "{channel_id}" from provider "{provider}"'}) - try: + def fetch_fn(): manifest_text, ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( provider, channel_id, manifest_url ) - rewriter = MPDRewriter( - self.media_proxy_url, - provider_proxy_url, - None, - highest_quality_only, - provider=provider, - channel=channel_id, + self.media_proxy_url, provider_proxy_url, None, highest_quality_only, + provider=provider, channel=channel_id, segment_headers=segment_headers, + ) + return rewriter.rewrite_mpd(manifest_text, effective_url), ttl + + return self._fetch_and_cache_manifest( + fetch_fn, + cache_set=lambda content, ttl: self.mpd_cache.set( + provider=provider, channel_id=channel_id, mpd_content=content, ttl=ttl, original_url=manifest_url + ), + cache_get_stale=lambda max_stale: self.mpd_cache.get(provider, channel_id, max_stale=max_stale), + cacheable=not highest_quality_only, + stale_window=15, + label=f"manifest for {provider}/{channel_id}", + ) + + def get_proxied_catchup_manifest(self, provider: str, channel_id: str, start_time: int, end_time: int, + epg_id: str = None, country: str = None) -> str: + cache_key = f"{channel_id}_catchup_{start_time}_{end_time}" + + cached_mpd = self.mpd_cache.get(provider, cache_key) + if cached_mpd: + response.content_type = "application/dash+xml; charset=utf-8" + return cached_mpd + + logger.info(f"Cache miss for catchup {provider}/{channel_id}, fetching manifest") + + if not self.media_proxy_url: + response.status = 503 + response.content_type = "application/json" + return json.dumps({"error": "Media proxy not configured (MEDIA_PROXY_URL not set)"}) + + manifest_url = self.manager.get_catchup_manifest( + provider_name=provider, channel_id=channel_id, start_time=start_time, end_time=end_time, + epg_id=epg_id, country=country, + ) + if not manifest_url: + response.status = 404 + response.content_type = "application/json" + return json.dumps({"error": "Catchup manifest not available"}) + + def fetch_fn(): + manifest_text, ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( + provider, channel_id, manifest_url + ) + rewriter = MPDRewriter( + self.media_proxy_url, provider_proxy_url, None, False, + provider=provider, channel=channel_id, segment_headers=segment_headers, + ) + return rewriter.rewrite_mpd(manifest_text, effective_url), ttl + + return self._fetch_and_cache_manifest( + fetch_fn, + cache_set=lambda content, ttl: self.mpd_cache.set( + provider=provider, channel_id=cache_key, mpd_content=content, ttl=ttl, original_url=manifest_url + ), + cache_get_stale=lambda max_stale: self.mpd_cache.get(provider, cache_key, max_stale=max_stale), + cacheable=True, + stale_window=15, + label=f"catchup manifest for {provider}/{channel_id}", + ) + + def get_decrypted_manifest(self, provider: str, channel_id: str, keyids: dict, + highest_quality_only: bool = False, receiver_side: bool = False, + drm_variant: str = "auto") -> str: + country = request.query.get("country") + cache_key = f"{provider}_{channel_id}_drm_{drm_variant}" + + if not highest_quality_only: + cached_mpd = self._get_decrypted_cached(cache_key) + if cached_mpd: + response.content_type = "application/dash+xml; charset=utf-8" + return cached_mpd + + logger.info( + f"Generating {'receiver-side clearkey' if receiver_side else 'decrypted'} manifest " + f"for {provider}/{channel_id} (highest_quality_only={highest_quality_only}, drm_variant={drm_variant})" + ) + + manifest_url = self.manager.get_channel_manifest( + provider_name=provider, channel_id=channel_id, country=country, drm_variant=drm_variant, + ) + if not manifest_url: + response.status = 404 + response.content_type = "application/json" + return json.dumps( + {"error": f'Manifest not available for channel "{channel_id}" from provider "{provider}"'}) + + def fetch_fn(): + manifest_text, ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( + provider, channel_id, manifest_url + ) + rewriter = MPDRewriter( + self.media_proxy_url, provider_proxy_url, keyids, highest_quality_only, + provider=provider, channel=channel_id, clearkey_receiver_side=receiver_side, segment_headers=segment_headers, ) rewritten_mpd = rewriter.rewrite_mpd(manifest_text, effective_url) + return rewritten_mpd, min(ttl, 10) # holds key material — keep exposure window short - if not highest_quality_only: - self.mpd_cache.set(provider=provider, channel_id=channel_id, - mpd_content=rewritten_mpd, ttl=ttl, original_url=manifest_url) + return self._fetch_and_cache_manifest( + fetch_fn, + cache_set=lambda content, ttl: self._set_decrypted_cached(cache_key, content, ttl), + cache_get_stale=lambda max_stale: self._get_decrypted_cached(cache_key, max_stale=max_stale), + cacheable=not highest_quality_only, + stale_window=5, + label=f"decrypted manifest for {provider}/{channel_id}", + ) - response.content_type = "application/dash+xml; charset=utf-8" - return rewritten_mpd + def get_decrypted_catchup_manifest(self, provider: str, channel_id: str, start_time: int, end_time: int, + keyids: dict, epg_id: str = None, + highest_quality_only: bool = False) -> str: + country = request.query.get("country") + cache_key = f"{provider}_{channel_id}_catchup_{start_time}_{end_time}_drm" - except ValueError as e: - logger.error(str(e)) - response.status = 502 + if not highest_quality_only: + cached_mpd = self._get_decrypted_cached(cache_key) + if cached_mpd: + response.content_type = "application/dash+xml; charset=utf-8" + return cached_mpd + + logger.info( + f"Generating decrypted catchup manifest for {provider}/{channel_id} " + f"(start={start_time} end={end_time} highest_quality_only={highest_quality_only})" + ) + + manifest_url = self.manager.get_catchup_manifest( + provider_name=provider, channel_id=channel_id, start_time=start_time, end_time=end_time, + epg_id=epg_id, country=country, + ) + if not manifest_url: + response.status = 404 response.content_type = "application/json" - return json.dumps({"error": str(e)}) - except Exception as fetch_err: - logger.error(f"Failed to fetch manifest: {fetch_err}") - response.status = 502 - response.content_type = "application/json" - return json.dumps({"error": f"Failed to fetch manifest: {str(fetch_err)}"}) + return json.dumps( + {"error": f'Catchup manifest not available for channel "{channel_id}" from provider "{provider}"'}) + + def fetch_fn(): + manifest_text, ttl, provider_proxy_url, segment_headers, effective_url = self.fetch_manifest_for_rewriter( + provider, channel_id, manifest_url + ) + rewriter = MPDRewriter( + self.media_proxy_url, provider_proxy_url, keyids, highest_quality_only, + provider=provider, channel=channel_id, clearkey_receiver_side=False, + segment_headers=segment_headers, + ) + rewritten_mpd = rewriter.rewrite_mpd(manifest_text, effective_url) + return rewritten_mpd, min(ttl, 30) + + return self._fetch_and_cache_manifest( + fetch_fn, + cache_set=lambda content, ttl: self._set_decrypted_cached(cache_key, content, ttl), + cache_get_stale=lambda max_stale: self._get_decrypted_cached(cache_key, max_stale=max_stale), + cacheable=not highest_quality_only, + stale_window=10, + label=f"decrypted catchup manifest for {provider}/{channel_id}", + ) @staticmethod def _sort_channels(channels):