From de3b0851416c3ac5ca0886dc6ce8f7e42ce357b1 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 12 Jun 2026 11:00:24 +0200 Subject: [PATCH] magenta2 event rewriter --- .idea/script.service.ultimate.iml | 1 + .../base/utils/mpd_event_rewriter.py | 379 ++++++++++++++++++ .../providers/magenta2/catchup_adjuster.py | 60 +++ routes/streams.py | 3 +- service.py | 20 +- 5 files changed, 458 insertions(+), 5 deletions(-) create mode 100644 lib/streaming_providers/base/utils/mpd_event_rewriter.py create mode 100644 lib/streaming_providers/providers/magenta2/catchup_adjuster.py diff --git a/.idea/script.service.ultimate.iml b/.idea/script.service.ultimate.iml index b06e1b3..295ce2c 100644 --- a/.idea/script.service.ultimate.iml +++ b/.idea/script.service.ultimate.iml @@ -3,6 +3,7 @@ + diff --git a/lib/streaming_providers/base/utils/mpd_event_rewriter.py b/lib/streaming_providers/base/utils/mpd_event_rewriter.py new file mode 100644 index 0000000..32cab63 --- /dev/null +++ b/lib/streaming_providers/base/utils/mpd_event_rewriter.py @@ -0,0 +1,379 @@ +# streaming_providers/base/utils/mpd_event_rewriter.py +""" +MPD Event Rewriter: shifts availabilityStartTime to a target EventStream event +so players start catchup playback at the correct programme boundary. + +Key design constraint +--------------------- +In a DASH dynamic MPD, Event.presentationTime is an *offset* (in timescale +ticks) from availabilityStartTime — it is NOT a Unix timestamp. All +wall-clock conversions must add availabilityStartTime to the offset. +""" + +import base64 +import xml.etree.ElementTree as ET +from datetime import datetime, timezone, timedelta +from typing import Optional, List, Dict, Tuple +from dataclasses import dataclass, field + +from .logger import logger + + +@dataclass +class EventInfo: + """One Event element from an MPD EventStream.""" + id: str + presentation_time: int # ticks from availabilityStartTime + duration: int # ticks + timescale: int + payload_base64: str + decoded_payload: Dict[str, str] = field(default_factory=dict) + title: Optional[str] = None + + def wall_clock_start(self, availability_start: datetime) -> datetime: + """Absolute UTC start time of this event.""" + return availability_start + timedelta(seconds=self.presentation_time / self.timescale) + + def wall_clock_end(self, availability_start: datetime) -> datetime: + """Absolute UTC end time of this event.""" + return availability_start + timedelta( + seconds=(self.presentation_time + self.duration) / self.timescale + ) + + def duration_seconds(self) -> float: + return self.duration / self.timescale + + +@dataclass(frozen=True) +class ExtractedEvents: + """ + Result of parsing EventStream data from an MPD. + + frozen=True prevents accidental reassignment of availability_start, + which would silently break all subsequent offset calculations. + """ + events: List[EventInfo] + availability_start: datetime + + +class MPDEventRewriter: + """ + Rewrites a dynamic DASH MPD so playback starts at a specific EventStream + event. + + Strategy + -------- + Set availabilityStartTime to the event's absolute wall-clock start time + while leaving all presentationTime values untouched. The player's + presentation clock then starts at time 0 == event start, so it seeks + directly to the right buffer position without any client-side adjustment. + """ + + NAMESPACES = { + 'mpd': 'urn:mpeg:dash:schema:mpd:2011', + 'cenc': 'urn:mpeg:cenc:2013', + 'xsi': 'http://www.w3.org/2001/XMLSchema-instance', + } + + EPG_SCHEME_URIS = { + "urn:de:dtag:eit:2017", + "urn:dvb:iptv:2014:eit", + } + + def __init__(self) -> None: + for prefix, uri in self.NAMESPACES.items(): + ET.register_namespace(prefix if prefix != 'mpd' else '', uri) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def extract_events( + self, + mpd_content: str, + scheme_id_uri: Optional[str] = None, + ) -> ExtractedEvents: + """ + Parse all EventStream events from *mpd_content*. + + Returns + ------- + ExtractedEvents + .events — list of EventInfo, may be empty + .availability_start — parsed availabilityStartTime (UTC) + + Raises + ------ + ValueError if availabilityStartTime is missing or unparseable. + """ + root = ET.fromstring(mpd_content) + availability_start = self._parse_availability_start(root) + events: List[EventInfo] = [] + + for es in root.findall('.//mpd:EventStream', self.NAMESPACES): + if scheme_id_uri and es.get('schemeIdUri', '') != scheme_id_uri: + continue + timescale = int(es.get('timescale', '10000000')) + + for ev in es.findall('mpd:Event', self.NAMESPACES): + info = EventInfo( + id=ev.get('id', ''), + presentation_time=int(ev.get('presentationTime', '0')), + duration=int(ev.get('duration', '0')), + timescale=timescale, + payload_base64=ev.text or '', + ) + if info.payload_base64: + info.decoded_payload = self._decode_epg_payload(info.payload_base64) + info.title = self._extract_title(info.decoded_payload) + + wall_start = info.wall_clock_start(availability_start) + logger.debug( + f"Event {info.id}: title={info.title!r}, " + f"wall_start={wall_start.isoformat()}, " + f"duration={info.duration_seconds():.0f}s" + ) + events.append(info) + + # Emit a single warning here rather than requiring every call site to + # remember to validate. The most common real-world cause is a timescale + # mismatch between the EventStream and the encoder, not genuinely bad data. + if events and all(e.presentation_time == 0 for e in events): + logger.warning( + "All events have presentationTime=0 — EventStream may be malformed " + "or there is a timescale mismatch; catchup alignment may be incorrect" + ) + + return ExtractedEvents(events=events, availability_start=availability_start) + + @staticmethod + def find_closest_event( + extracted: ExtractedEvents, + target_time: datetime, + ) -> Tuple[Optional[EventInfo], float]: + """ + Return (best_event, diff_seconds) where diff is the absolute distance + between target_time and the event's wall-clock start. + """ + if not extracted.events: + return None, float('inf') + + best: Optional[EventInfo] = None + best_diff = float('inf') + + for ev in extracted.events: + diff = abs( + (ev.wall_clock_start(extracted.availability_start) - target_time).total_seconds() + ) + if diff < best_diff: + best_diff = diff + best = ev + + if best: + logger.debug( + f"Closest event to {target_time.isoformat()}: " + f"{best.title!r} (diff={best_diff:.0f}s)" + ) + return best, best_diff + + def rewrite_for_event( + self, + mpd_content: str, + extracted: ExtractedEvents, + target_event: EventInfo, + keep_other_events: bool = False, + force_static_if_ended: bool = True, + ) -> str: + """ + Rewrite *mpd_content* so the player starts at *target_event*. + + Sets availabilityStartTime = target_event.wall_clock_start() and + preserves all presentationTime values. Optionally removes other + events from the EventStream to avoid confusing the player. + """ + root = ET.fromstring(mpd_content) + event_wall_start = target_event.wall_clock_start(extracted.availability_start) + + # --- Update MPD root attributes --- + ast_str = event_wall_start.strftime('%Y-%m-%dT%H:%M:%S') + 'Z' + root.set('availabilityStartTime', ast_str) + root.set('publishTime', ast_str) + root.set('suggestedPresentationDelay', 'PT0S') + + now = datetime.now(tz=timezone.utc) + event_ended = now > target_event.wall_clock_end(extracted.availability_start) + + if force_static_if_ended and event_ended: + root.set('type', 'static') + root.set( + 'mediaPresentationDuration', + self._seconds_to_duration(target_event.duration_seconds()), + ) + logger.info( + f"Event {target_event.id} ended — converting to static MPD, " + f"duration={self._seconds_to_duration(target_event.duration_seconds())}" + ) + else: + root.set('type', 'dynamic') + + # --- Filter EventStream --- + for es in root.findall('.//mpd:EventStream', self.NAMESPACES): + if keep_other_events: + continue + for ev in list(es.findall('mpd:Event', self.NAMESPACES)): + es.remove(ev) + self._append_event(es, target_event) + + return self._serialise(root) + + def rewrite_by_buffer_offset( + self, + mpd_content: str, + extracted: ExtractedEvents, + requested_start: datetime, + max_buffer_seconds: int = 14400, + ) -> str: + """ + Fallback: shift availabilityStartTime so the requested_start maps to + presentation time 0. + + availabilityStartTime is set directly to requested_start (clamped to + the buffer window). The player will seek to the head of the live + window which corresponds to the requested time. + + timeShiftBufferDepth is intentionally left at the provider's declared + value (the full DVR window). Adjusting it to the offset would shrink + the seekable range to exactly the catchup point, which is the opposite + of what a catchup UI wants. Players that honour it (e.g. + inputstream.adaptive) use the segment timeline for seek range anyway. + """ + now = datetime.now(tz=timezone.utc) + offset = (now - requested_start).total_seconds() + + if offset > max_buffer_seconds: + logger.warning( + f"Requested time {requested_start.isoformat()} is outside " + f"{max_buffer_seconds // 3600}h buffer; clamping." + ) + requested_start = now - timedelta(seconds=max_buffer_seconds) + + root = ET.fromstring(mpd_content) + ast_str = requested_start.strftime('%Y-%m-%dT%H:%M:%S') + 'Z' + root.set('availabilityStartTime', ast_str) + root.set('publishTime', ast_str) + root.set('suggestedPresentationDelay', 'PT0S') + root.set('type', 'dynamic') + + logger.info( + f"Buffer-offset rewrite: AST {extracted.availability_start.isoformat()} " + f"→ {requested_start.isoformat()} (offset={offset:.0f}s)" + ) + return self._serialise(root) + + def get_event_timeline( + self, + mpd_content: str, + scheme_id_uri: Optional[str] = None, + ) -> List[Dict]: + """Human-readable event list for debugging.""" + extracted = self.extract_events(mpd_content, scheme_id_uri) + timeline = [] + for ev in extracted.events: + timeline.append({ + 'id': ev.id, + 'title': ev.title, + 'wall_start': ev.wall_clock_start(extracted.availability_start).isoformat(), + 'wall_end': ev.wall_clock_end(extracted.availability_start).isoformat(), + 'duration_seconds': ev.duration_seconds(), + 'presentation_time_ticks': ev.presentation_time, + 'timescale': ev.timescale, + }) + return sorted(timeline, key=lambda x: x['presentation_time_ticks']) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + @staticmethod + def _parse_availability_start(root: ET.Element) -> datetime: + raw = root.get('availabilityStartTime', '') + if not raw: + raise ValueError("MPD missing availabilityStartTime") + try: + return datetime.fromisoformat(raw.replace('Z', '+00:00')) + except ValueError: + raise ValueError(f"Cannot parse availabilityStartTime: {raw!r}") + + @staticmethod + def _decode_epg_payload(payload_b64: str) -> Dict[str, str]: + try: + decoded = base64.b64decode(payload_b64).decode('utf-8', errors='ignore') + result: Dict[str, str] = {} + for part in decoded.split(','): + if '=' not in part: + continue + key, value = part.split('=', 1) + if key == 'DES_4D': + try: + result[key] = ( + bytes.fromhex(value) + .replace(b'\x00', b'') + .decode('utf-8', errors='ignore') + ) + except (ValueError, TypeError): + result[key] = value + else: + result[key] = value + return result + except Exception as e: + logger.debug(f"EPG payload decode failed: {e}") + return {} + + @staticmethod + def _extract_title(payload: Dict[str, str]) -> Optional[str]: + title = payload.get('DES_4D') or payload.get('TITLE') or payload.get('NAME') + if title: + title = ' '.join(title.split()) + if len(title) > 100: + title = title[:97] + '...' + return title or None + + @staticmethod + def _append_event(es: ET.Element, ev: EventInfo) -> None: + elem = ET.Element('Event', { + 'presentationTime': str(ev.presentation_time), + 'duration': str(ev.duration), + 'id': ev.id, + 'contentEncoding': 'base64', + }) + elem.text = ev.payload_base64 + es.append(elem) + + @staticmethod + def _seconds_to_duration(seconds: float) -> str: + s = int(round(seconds)) + h, rem = divmod(s, 3600) + m, sec = divmod(rem, 60) + if h: + return f"PT{h}H{m}M{sec}S" + if m: + return f"PT{m}M{sec}S" + return f"PT{sec}S" + + @staticmethod + def _serialise(root: ET.Element) -> str: + text = ET.tostring(root, encoding='unicode', method='xml') + if not text.startswith('\n' + text + return text + + +# Module-level singleton +_instance: Optional[MPDEventRewriter] = None + + +def get_mpd_event_rewriter() -> MPDEventRewriter: + global _instance + if _instance is None: + _instance = MPDEventRewriter() + return _instance \ No newline at end of file diff --git a/lib/streaming_providers/providers/magenta2/catchup_adjuster.py b/lib/streaming_providers/providers/magenta2/catchup_adjuster.py new file mode 100644 index 0000000..ae5ccf5 --- /dev/null +++ b/lib/streaming_providers/providers/magenta2/catchup_adjuster.py @@ -0,0 +1,60 @@ +# streaming_providers/magenta2/catchup_adjuster.py +""" +Magenta2 catchup manifest adjuster. +""" + +from datetime import datetime, timezone +from ...base.utils.logger import logger +from ...base.utils.mpd_event_rewriter import get_mpd_event_rewriter + +_MAGENTA2_EPG_SCHEME = "urn:de:dtag:eit:2017" +_EVENT_MATCH_THRESHOLD_S = 300 + + +class Magenta2CatchupAdjuster: + """Adjusts Magenta2 catchup manifests to requested start time.""" + + @staticmethod + def adjust(mpd_content: str, requested_start_time: int) -> str: + """ + Adjust Magenta2 catchup MPD to start at requested time. + + Returns adjusted MPD content, or original if adjustment fails. + """ + rewriter = get_mpd_event_rewriter() + requested_dt = datetime.fromtimestamp(requested_start_time, tz=timezone.utc) + + try: + events, ast = rewriter.extract_events(mpd_content, _MAGENTA2_EPG_SCHEME) + + if not events: + logger.warning("Magenta2: no events found, using buffer offset") + return rewriter.rewrite_by_buffer_offset(mpd_content, ast, requested_dt) + + best_event, diff = rewriter.find_closest_event(events, ast, requested_dt) + + if best_event and diff <= _EVENT_MATCH_THRESHOLD_S: + logger.info( + f"Magenta2: adjusting to event '{best_event.title}' " + f"(start={best_event.wall_clock_start(ast).isoformat()}, diff={diff:.0f}s)" + ) + return rewriter.rewrite_for_event( + mpd_content, ast, best_event, + keep_other_events=False, + force_static_if_ended=False, + ) + + logger.info(f"Magenta2: no close event match (diff={diff:.0f}s), using buffer offset") + return rewriter.rewrite_by_buffer_offset(mpd_content, ast, requested_dt) + + except ValueError as e: + logger.warning(f"Magenta2: cannot parse events ({e}), using buffer offset") + try: + _, ast = rewriter.extract_events(mpd_content) # fallback without scheme + return rewriter.rewrite_by_buffer_offset(mpd_content, ast, requested_dt) + except ValueError: + logger.error("Magenta2: MPD has no availabilityStartTime") + return mpd_content + except Exception as e: + logger.error(f"Magenta2: adjustment failed: {e}") + return mpd_content \ No newline at end of file diff --git a/routes/streams.py b/routes/streams.py index 779bca9..2d4e8b0 100644 --- a/routes/streams.py +++ b/routes/streams.py @@ -347,7 +347,8 @@ def setup_stream_routes(app, manager, service): if is_catchup: if manager.needs_proxy(provider): return service.get_proxied_catchup_manifest( - provider, content_id, start_time, end_time, epg_id, country + provider, content_id, start_time, end_time, epg_id, country, + drm_variant=drm_variant, ) else: manifest_url = manager.get_catchup_manifest( diff --git a/service.py b/service.py index 2b67113..87069b2 100644 --- a/service.py +++ b/service.py @@ -474,12 +474,13 @@ class UltimateService: end_time: int, epg_id: str = None, country: str = None, + drm_variant: str = "auto", ) -> 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}" + # Use start_time for cache key (requested_start_time is same value, but start_time is guaranteed) + cache_key = f"{channel_id}_catchup_start_{start_time}" if provider.lower() == "magenta2" else f"{channel_id}_catchup_{start_time}_{end_time}" cached_mpd = self.mpd_cache.get(provider, cache_key) if cached_mpd: @@ -511,11 +512,22 @@ class UltimateService: provider, channel_id, manifest_url ) + # ==================================================================== + # MAGENTA2 TIMELINE ADJUSTMENT + # ==================================================================== + if provider.lower() == "magenta2" and start_time: + from streaming_providers.providers.magenta2.catchup_adjuster import Magenta2CatchupAdjuster + manifest_text = Magenta2CatchupAdjuster.adjust(manifest_text, start_time) + ttl = min(ttl, 300) # Shorter TTL for adjusted manifests + + # ==================================================================== + # STANDARD PROXY REWRITING + # ==================================================================== rewriter = MPDRewriter( self.media_proxy_url, provider_proxy_url, - None, # No keyids for catchup streams - False, # highest_quality_only — not needed for catchup + None, + False, provider=provider, channel=channel_id, segment_headers=segment_headers,