import logging import math import re from datetime import datetime, timedelta, timezone from typing import List, Dict, Optional, Union from urllib.parse import urljoin import xmltodict logger = logging.getLogger(__name__) def resolve_url(base_url: str, relative_url: str) -> str: """ Resolve a relative URL against a base URL. Handles three cases: 1. Absolute URL (starts with http:// or https://) - return as-is 2. Absolute path (starts with /) - resolve against origin (scheme + host) 3. Relative path - resolve against base URL directory Args: base_url: The base URL (typically the MPD URL) relative_url: The URL to resolve Returns: The resolved absolute URL """ if not relative_url: return base_url # Already absolute URL if relative_url.startswith(("http://", "https://")): return relative_url # Use urljoin which correctly handles: # - Absolute paths (starting with /) -> resolves against origin # - Relative paths -> resolves against base URL return urljoin(base_url, relative_url) def parse_mpd(mpd_content: Union[str, bytes]) -> dict: """ Parses the MPD content into a dictionary. Args: mpd_content (Union[str, bytes]): The MPD content to parse. Returns: dict: The parsed MPD content as a dictionary. """ return xmltodict.parse(mpd_content) def parse_mpd_dict( mpd_dict: dict, mpd_url: str, parse_drm: bool = True, parse_segment_profile_id: Optional[str] = None ) -> dict: """ Parses the MPD dictionary and extracts relevant information. Args: mpd_dict (dict): The MPD content as a dictionary. mpd_url (str): The URL of the MPD manifest. parse_drm (bool, optional): Whether to parse DRM information. Defaults to True. parse_segment_profile_id (str, optional): The profile ID to parse segments for. Defaults to None. Returns: dict: The parsed MPD information including profiles and DRM info. This function processes the MPD dictionary to extract profiles, DRM information, and other relevant data. It handles both live and static MPD manifests. """ profiles = [] parsed_dict = {} is_live = mpd_dict["MPD"].get("@type", "static").lower() == "dynamic" parsed_dict["isLive"] = is_live media_presentation_duration = mpd_dict["MPD"].get("@mediaPresentationDuration") # Parse additional MPD attributes for live streams if is_live: parsed_dict["minimumUpdatePeriod"] = parse_duration(mpd_dict["MPD"].get("@minimumUpdatePeriod", "PT0S")) parsed_dict["timeShiftBufferDepth"] = parse_duration(mpd_dict["MPD"].get("@timeShiftBufferDepth", "PT2M")) parsed_dict["availabilityStartTime"] = datetime.fromisoformat( mpd_dict["MPD"]["@availabilityStartTime"].replace("Z", "+00:00") ) parsed_dict["publishTime"] = datetime.fromisoformat( mpd_dict["MPD"].get("@publishTime", "").replace("Z", "+00:00") ) periods = mpd_dict["MPD"]["Period"] periods = periods if isinstance(periods, list) else [periods] for period in periods: parsed_dict["PeriodStart"] = parse_duration(period.get("@start", "PT0S")) adaptation_sets = period["AdaptationSet"] adaptation_sets = adaptation_sets if isinstance(adaptation_sets, list) else [adaptation_sets] for adaptation in adaptation_sets: representations = adaptation["Representation"] representations = representations if isinstance(representations, list) else [representations] for representation in representations: profile = parse_representation( parsed_dict, representation, adaptation, mpd_url, media_presentation_duration, parse_segment_profile_id, ) if profile: profiles.append(profile) parsed_dict["profiles"] = profiles if parse_drm: drm_info = extract_drm_info(periods, mpd_url) else: drm_info = {} parsed_dict["drmInfo"] = drm_info return parsed_dict def pad_base64(encoded_key_id): """ Pads a base64 encoded key ID to make its length a multiple of 4. Args: encoded_key_id (str): The base64 encoded key ID. Returns: str: The padded base64 encoded key ID. """ return encoded_key_id + "=" * (4 - len(encoded_key_id) % 4) def extract_drm_info(periods: List[Dict], mpd_url: str) -> Dict: """ Extracts DRM information from the MPD periods. Args: periods (List[Dict]): The list of periods in the MPD. mpd_url (str): The URL of the MPD manifest. Returns: Dict: The extracted DRM information. This function processes the ContentProtection elements in the MPD to extract DRM system information, such as ClearKey, Widevine, and PlayReady. """ drm_info = {"isDrmProtected": False} for period in periods: adaptation_sets: Union[list[dict], dict] = period.get("AdaptationSet", []) if not isinstance(adaptation_sets, list): adaptation_sets = [adaptation_sets] for adaptation_set in adaptation_sets: # Check ContentProtection in AdaptationSet process_content_protection(adaptation_set.get("ContentProtection", []), drm_info) # Check ContentProtection inside each Representation representations: Union[list[dict], dict] = adaptation_set.get("Representation", []) if not isinstance(representations, list): representations = [representations] for representation in representations: process_content_protection(representation.get("ContentProtection", []), drm_info) # If we have a license acquisition URL, make sure it's absolute if "laUrl" in drm_info and not drm_info["laUrl"].startswith(("http://", "https://")): drm_info["laUrl"] = urljoin(mpd_url, drm_info["laUrl"]) return drm_info def process_content_protection(content_protection: Union[list[dict], dict], drm_info: dict): """ Processes the ContentProtection elements to extract DRM information. Args: content_protection (Union[list[dict], dict]): The ContentProtection elements. drm_info (dict): The dictionary to store DRM information. This function updates the drm_info dictionary with DRM system information found in the ContentProtection elements. """ if not isinstance(content_protection, list): content_protection = [content_protection] for protection in content_protection: drm_info["isDrmProtected"] = True scheme_id_uri = protection.get("@schemeIdUri", "").lower() if "clearkey" in scheme_id_uri: drm_info["drmSystem"] = "clearkey" if "clearkey:Laurl" in protection: la_url = protection["clearkey:Laurl"].get("#text") if la_url and "laUrl" not in drm_info: drm_info["laUrl"] = la_url elif "widevine" in scheme_id_uri or "edef8ba9-79d6-4ace-a3c8-27dcd51d21ed" in scheme_id_uri: drm_info["drmSystem"] = "widevine" pssh = protection.get("cenc:pssh", {}).get("#text") if pssh: drm_info["pssh"] = pssh elif "playready" in scheme_id_uri or "9a04f079-9840-4286-ab92-e65be0885f95" in scheme_id_uri: drm_info["drmSystem"] = "playready" if "@cenc:default_KID" in protection: key_id = protection["@cenc:default_KID"].replace("-", "") if "keyId" not in drm_info: drm_info["keyId"] = key_id if "ms:laurl" in protection: la_url = protection["ms:laurl"].get("@licenseUrl") if la_url and "laUrl" not in drm_info: drm_info["laUrl"] = la_url return drm_info def parse_representation( parsed_dict: dict, representation: dict, adaptation: dict, mpd_url: str, media_presentation_duration: str, parse_segment_profile_id: Optional[str], ) -> Optional[dict]: """ Parses a representation and extracts profile information. Args: parsed_dict (dict): The parsed MPD data. representation (dict): The representation data. adaptation (dict): The adaptation set data. mpd_url (str): The URL of the MPD manifest. media_presentation_duration (str): The media presentation duration. parse_segment_profile_id (str, optional): The profile ID to parse segments for. Defaults to None. Returns: Optional[dict]: The parsed profile information or None if not applicable. """ mime_type = _get_key(adaptation, representation, "@mimeType") or ( "video/mp4" if "avc" in representation["@codecs"] else "audio/mp4" ) if "video" not in mime_type and "audio" not in mime_type: return None profile = { "id": representation.get("@id") or adaptation.get("@id"), "mimeType": mime_type, "lang": representation.get("@lang") or adaptation.get("@lang"), "codecs": representation.get("@codecs") or adaptation.get("@codecs"), "bandwidth": int(representation.get("@bandwidth") or adaptation.get("@bandwidth")), "startWithSAP": (_get_key(adaptation, representation, "@startWithSAP") or "1") == "1", "mediaPresentationDuration": media_presentation_duration, } if "audio" in profile["mimeType"]: profile["audioSamplingRate"] = representation.get("@audioSamplingRate") or adaptation.get("@audioSamplingRate") profile["channels"] = representation.get("AudioChannelConfiguration", {}).get("@value", "2") else: # Handle video-specific attributes, making them optional with sensible defaults if "@width" in representation: profile["width"] = int(representation["@width"]) elif "@width" in adaptation: profile["width"] = int(adaptation["@width"]) else: profile["width"] = 0 # Default if width is missing if "@height" in representation: profile["height"] = int(representation["@height"]) elif "@height" in adaptation: profile["height"] = int(adaptation["@height"]) else: profile["height"] = 0 # Default if height is missing frame_rate = representation.get("@frameRate") or adaptation.get("@maxFrameRate") or "30000/1001" frame_rate = frame_rate if "/" in frame_rate else f"{frame_rate}/1" profile["frameRate"] = round(int(frame_rate.split("/")[0]) / int(frame_rate.split("/")[1]), 3) profile["sar"] = representation.get("@sar", "1:1") # Extract segment template start number for adaptive sequence calculation segment_template_data = adaptation.get("SegmentTemplate") or representation.get("SegmentTemplate") if segment_template_data: try: profile["segment_template_start_number"] = int(segment_template_data.get("@startNumber", 1)) except (ValueError, TypeError): profile["segment_template_start_number"] = 1 else: profile["segment_template_start_number"] = 1 # For SegmentBase profiles, we need to set initUrl even when not parsing segments # This is needed for the HLS playlist builder to reference the init URL segment_base_data = representation.get("SegmentBase") if segment_base_data and "initUrl" not in profile: base_url = representation.get("BaseURL", "") profile["initUrl"] = resolve_url(mpd_url, base_url) # Store initialization range if available if "Initialization" in segment_base_data: init_range = segment_base_data["Initialization"].get("@range") if init_range: profile["initRange"] = init_range # For SegmentList profiles, we also need to set initUrl even when not parsing segments segment_list_data = representation.get("SegmentList") or adaptation.get("SegmentList") if segment_list_data and "initUrl" not in profile: if "Initialization" in segment_list_data: init_data = segment_list_data["Initialization"] if "@sourceURL" in init_data: init_url = init_data["@sourceURL"] profile["initUrl"] = resolve_url(mpd_url, init_url) elif "@range" in init_data: base_url = representation.get("BaseURL", "") profile["initUrl"] = resolve_url(mpd_url, base_url) profile["initRange"] = init_data["@range"] if parse_segment_profile_id is None or profile["id"] != parse_segment_profile_id: return profile # Parse segments based on the addressing scheme used segment_template = adaptation.get("SegmentTemplate") or representation.get("SegmentTemplate") segment_list = adaptation.get("SegmentList") or representation.get("SegmentList") # Get BaseURL from representation (can be relative path like "a/b/c/") base_url = representation.get("BaseURL", "") if segment_template: profile["segments"] = parse_segment_template(parsed_dict, segment_template, profile, mpd_url, base_url) elif segment_list: # Get timescale from SegmentList or default to 1 timescale = int(segment_list.get("@timescale", 1)) profile["segments"] = parse_segment_list(adaptation, representation, profile, mpd_url, timescale) else: profile["segments"] = parse_segment_base(representation, profile, mpd_url) return profile def _get_key(adaptation: dict, representation: dict, key: str) -> Optional[str]: """ Retrieves a key from the representation or adaptation set. Args: adaptation (dict): The adaptation set data. representation (dict): The representation data. key (str): The key to retrieve. Returns: Optional[str]: The value of the key or None if not found. """ return representation.get(key, adaptation.get(key, None)) def parse_segment_template( parsed_dict: dict, item: dict, profile: dict, mpd_url: str, base_url: str = "" ) -> List[Dict]: """ Parses a segment template and extracts segment information. Args: parsed_dict (dict): The parsed MPD data. item (dict): The segment template data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. base_url (str): The BaseURL from the representation (optional, for per-representation paths). Returns: List[Dict]: The list of parsed segments. """ segments = [] timescale = int(item.get("@timescale", 1)) # Initialization if "@initialization" in item: media = item["@initialization"] media = media.replace("$RepresentationID$", profile["id"]) media = media.replace("$Bandwidth$", str(profile["bandwidth"])) # Combine base_url and media, then resolve against mpd_url if base_url: media = base_url + media profile["initUrl"] = resolve_url(mpd_url, media) # Segments if "SegmentTimeline" in item: segments.extend(parse_segment_timeline(parsed_dict, item, profile, mpd_url, timescale, base_url)) elif "@duration" in item: segments.extend(parse_segment_duration(parsed_dict, item, profile, mpd_url, timescale, base_url)) return segments def parse_segment_timeline( parsed_dict: dict, item: dict, profile: dict, mpd_url: str, timescale: int, base_url: str = "" ) -> List[Dict]: """ Parses a segment timeline and extracts segment information. Args: parsed_dict (dict): The parsed MPD data. item (dict): The segment timeline data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. timescale (int): The timescale for the segments. base_url (str): The BaseURL from the representation (optional, for per-representation paths). Returns: List[Dict]: The list of parsed segments. """ timelines = item["SegmentTimeline"]["S"] timelines = timelines if isinstance(timelines, list) else [timelines] period_start = parsed_dict.get("availabilityStartTime", datetime.fromtimestamp(0, tz=timezone.utc)) + timedelta( seconds=parsed_dict.get("PeriodStart", 0) ) presentation_time_offset = int(item.get("@presentationTimeOffset", 0)) start_number = int(item.get("@startNumber", 1)) segments = [ create_segment_data(timeline, item, profile, mpd_url, timescale, base_url) for timeline in preprocess_timeline(timelines, start_number, period_start, presentation_time_offset, timescale) ] return segments def preprocess_timeline( timelines: List[Dict], start_number: int, period_start: datetime, presentation_time_offset: int, timescale: int ) -> List[Dict]: """ Preprocesses the segment timeline data. Args: timelines (List[Dict]): The list of timeline segments. start_number (int): The starting segment number. period_start (datetime): The start time of the period. presentation_time_offset (int): The presentation time offset. timescale (int): The timescale for the segments. Returns: List[Dict]: The list of preprocessed timeline segments. """ processed_data = [] current_time = 0 for timeline in timelines: repeat = int(timeline.get("@r", 0)) duration = int(timeline["@d"]) start_time = int(timeline.get("@t", current_time)) for _ in range(repeat + 1): segment_start_time = period_start + timedelta(seconds=(start_time - presentation_time_offset) / timescale) segment_end_time = segment_start_time + timedelta(seconds=duration / timescale) processed_data.append( { "number": start_number, "start_time": segment_start_time, "end_time": segment_end_time, "duration": duration, "time": start_time, "duration_mpd_timescale": duration, } ) start_time += duration start_number += 1 current_time = start_time return processed_data def parse_segment_duration( parsed_dict: dict, item: dict, profile: dict, mpd_url: str, timescale: int, base_url: str = "" ) -> List[Dict]: """ Parses segment duration and extracts segment information. This is used for static or live MPD manifests. Args: parsed_dict (dict): The parsed MPD data. item (dict): The segment duration data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. timescale (int): The timescale for the segments. base_url (str): The BaseURL from the representation (optional, for per-representation paths). Returns: List[Dict]: The list of parsed segments. """ duration = int(item["@duration"]) start_number = int(item.get("@startNumber", 1)) segment_duration_sec = duration / timescale if parsed_dict["isLive"]: segments = generate_live_segments(parsed_dict, segment_duration_sec, start_number) else: segments = generate_vod_segments(profile, duration, timescale, start_number) return [create_segment_data(seg, item, profile, mpd_url, timescale, base_url) for seg in segments] def generate_live_segments(parsed_dict: dict, segment_duration_sec: float, start_number: int) -> List[Dict]: """ Generates live segments based on the segment duration and start number. This is used for live MPD manifests. Args: parsed_dict (dict): The parsed MPD data. segment_duration_sec (float): The segment duration in seconds. start_number (int): The starting segment number. Returns: List[Dict]: The list of generated live segments. """ time_shift_buffer_depth = timedelta(seconds=parsed_dict.get("timeShiftBufferDepth", 60)) segment_count = math.ceil(time_shift_buffer_depth.total_seconds() / segment_duration_sec) current_time = datetime.now(tz=timezone.utc) earliest_segment_number = max( start_number + math.floor((current_time - parsed_dict["availabilityStartTime"]).total_seconds() / segment_duration_sec) - segment_count, start_number, ) return [ { "number": number, "start_time": parsed_dict["availabilityStartTime"] + timedelta(seconds=(number - start_number) * segment_duration_sec), "duration": segment_duration_sec, } for number in range(earliest_segment_number, earliest_segment_number + segment_count) ] def generate_vod_segments(profile: dict, duration: int, timescale: int, start_number: int) -> List[Dict]: """ Generates VOD segments based on the segment duration and start number. This is used for static MPD manifests. Args: profile (dict): The profile information. duration (int): The segment duration. timescale (int): The timescale for the segments. start_number (int): The starting segment number. Returns: List[Dict]: The list of generated VOD segments. """ total_duration = profile.get("mediaPresentationDuration") or 0 if isinstance(total_duration, str): total_duration = parse_duration(total_duration) segment_count = math.ceil(total_duration * timescale / duration) return [{"number": start_number + i, "duration": duration / timescale} for i in range(segment_count)] def create_segment_data( segment: Dict, item: dict, profile: dict, mpd_url: str, timescale: Optional[int] = None, base_url: str = "" ) -> Dict: """ Creates segment data based on the segment information. This includes the segment URL and metadata. Args: segment (Dict): The segment information. item (dict): The segment template data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. timescale (int, optional): The timescale for the segments. Defaults to None. base_url (str): The BaseURL from the representation (optional, for per-representation paths). Returns: Dict: The created segment data. """ media_template = item["@media"] media = media_template.replace("$RepresentationID$", profile["id"]) media = media.replace("$Number%04d$", f"{segment['number']:04d}") media = media.replace("$Number$", str(segment["number"])) media = media.replace("$Bandwidth$", str(profile["bandwidth"])) if "time" in segment and timescale is not None: media = media.replace("$Time$", str(int(segment["time"]))) # Combine base_url and media, then resolve against mpd_url if base_url: media = base_url + media media = resolve_url(mpd_url, media) segment_data = { "type": "segment", "media": media, "number": segment["number"], } # Add time and duration metadata for adaptive sequence calculation if "time" in segment: segment_data["time"] = segment["time"] if "duration" in segment: segment_data["duration_mpd_timescale"] = segment["duration"] if "start_time" in segment and "end_time" in segment: segment_data.update( { "start_time": segment["start_time"], "end_time": segment["end_time"], "extinf": (segment["end_time"] - segment["start_time"]).total_seconds(), "program_date_time": segment["start_time"].isoformat() + "Z", } ) elif "start_time" in segment and "duration" in segment: # duration here is in timescale units (from timeline segments) duration_seconds = segment["duration"] / timescale if timescale else segment["duration"] segment_data.update( { "start_time": segment["start_time"], "end_time": segment["start_time"] + timedelta(seconds=duration_seconds), "extinf": duration_seconds, "program_date_time": segment["start_time"].isoformat() + "Z", } ) elif "duration" in segment: # duration from generate_vod_segments and generate_live_segments is already in seconds segment_data["extinf"] = segment["duration"] return segment_data def parse_segment_list( adaptation: dict, representation: dict, profile: dict, mpd_url: str, timescale: int ) -> List[Dict]: """ Parses SegmentList element with explicit SegmentURL entries. SegmentList MPDs explicitly list each segment URL, unlike SegmentTemplate which uses URL patterns. This is less common but used by some packagers. Args: adaptation (dict): The adaptation set data. representation (dict): The representation data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. timescale (int): The timescale for duration calculations. Returns: List[Dict]: The list of parsed segments. """ # SegmentList can be at AdaptationSet or Representation level segment_list = representation.get("SegmentList") or adaptation.get("SegmentList", {}) segments = [] # Handle Initialization element if "Initialization" in segment_list: init_data = segment_list["Initialization"] if "@sourceURL" in init_data: init_url = init_data["@sourceURL"] profile["initUrl"] = resolve_url(mpd_url, init_url) elif "@range" in init_data: # Initialization by byte range on the BaseURL base_url = representation.get("BaseURL", "") profile["initUrl"] = resolve_url(mpd_url, base_url) profile["initRange"] = init_data["@range"] # Get segment duration from SegmentList attributes duration = int(segment_list.get("@duration", 0)) list_timescale = int(segment_list.get("@timescale", timescale or 1)) segment_duration_sec = duration / list_timescale if list_timescale else 0 # Parse SegmentURL elements segment_urls = segment_list.get("SegmentURL", []) if not isinstance(segment_urls, list): segment_urls = [segment_urls] for i, seg_url in enumerate(segment_urls): if seg_url is None: continue # Get media URL - can be @media attribute or use BaseURL with @mediaRange media_url = seg_url.get("@media", "") media_range = seg_url.get("@mediaRange") if media_url: media_url = resolve_url(mpd_url, media_url) else: # Use BaseURL with byte range base_url = representation.get("BaseURL", "") media_url = resolve_url(mpd_url, base_url) segment_data = { "type": "segment", "media": media_url, "number": i + 1, "extinf": segment_duration_sec if segment_duration_sec > 0 else 1.0, } # Include media range if specified if media_range: segment_data["mediaRange"] = media_range segments.append(segment_data) return segments def parse_segment_base(representation: dict, profile: dict, mpd_url: str) -> List[Dict]: """ Parses segment base information and extracts segment data. This is used for single-segment representations (SegmentBase MPDs, typically GPAC-generated on-demand profiles). For SegmentBase, the entire media file is treated as a single segment. The initialization data is specified by the Initialization element's range, and the segment index (SIDX) is at indexRange. Args: representation (dict): The representation data. profile (dict): The profile information. mpd_url (str): The URL of the MPD manifest. Returns: List[Dict]: The list of parsed segments. """ segment = representation.get("SegmentBase", {}) base_url = representation.get("BaseURL", "") # Build the full media URL media_url = resolve_url(mpd_url, base_url) # Set initUrl for SegmentBase - this is the URL with the initialization range # The initialization segment contains codec/track info needed before playing media profile["initUrl"] = media_url # For SegmentBase, we need to specify byte ranges for init and media segments init_range = None if "Initialization" in segment: init_range = segment["Initialization"].get("@range") # Store initialization range in profile for segment endpoint to use if init_range: profile["initRange"] = init_range # Get the index range which points to SIDX box index_range = segment.get("@indexRange", "") # Calculate total duration from profile's mediaPresentationDuration total_duration = profile.get("mediaPresentationDuration") if isinstance(total_duration, str): total_duration = parse_duration(total_duration) elif total_duration is None: total_duration = 0 # For SegmentBase, we return a single segment representing the entire media # The media URL is the same as initUrl but will be accessed with different byte ranges return [ { "type": "segment", "media": media_url, "number": 1, "extinf": total_duration if total_duration > 0 else 1.0, "indexRange": index_range, "initRange": init_range, } ] def parse_duration(duration_str: str) -> float: """ Parses a duration ISO 8601 string into seconds. Args: duration_str (str): The duration string to parse. Returns: float: The parsed duration in seconds. """ pattern = re.compile(r"P(?:(\d+)Y)?(?:(\d+)M)?(?:(\d+)D)?T?(?:(\d+)H)?(?:(\d+)M)?(?:(\d+(?:\.\d+)?)S)?") match = pattern.match(duration_str) if not match: raise ValueError(f"Invalid duration format: {duration_str}") years, months, days, hours, minutes, seconds = [float(g) if g else 0 for g in match.groups()] return years * 365 * 24 * 3600 + months * 30 * 24 * 3600 + days * 24 * 3600 + hours * 3600 + minutes * 60 + seconds