from __future__ import annotations import re import itertools import base64 import queue import subprocess import tempfile import os from collections import defaultdict from contextlib import suppress from typing import List, Self from datetime import timedelta from streamlink.exceptions import PluginError, FatalPluginError from streamlink.plugin import pluginmatcher, pluginargument from streamlink.plugin.plugin import (HIGH_PRIORITY, parse_params, stream_weight) from streamlink.stream.dash import (DASHStream, DASHStreamWorker, DASHStreamWriter, DASHStreamReader) from streamlink.stream.dash.manifest import MPD, Representation from streamlink.stream.dash.segment import DASHSegment from streamlink.stream.ffmpegmux import FFMPEGMuxer from streamlink.utils.url import update_scheme from streamlink.session import Streamlink from streamlink.utils.l10n import Language from streamlink.utils.times import fromtimestamp, now from streamlink.logger import getLogger from streamlink.plugins.dash import MPEGDASH from streamlink.utils.parse import parse_xml from typing import Any from collections.abc import Mapping period_sync_queue = queue.Queue() rep_sync_queue = queue.Queue() log = getLogger(__name__) class _DecryptedResponse: """ Wraps decrypted bytes to mimic the requests.Response interface expected by DASHStreamWriter.write() (which calls .iter_content()). """ def __init__(self, data: bytes): self._data = data def iter_content(self, chunk_size: int = 8192): for i in range(0, len(self._data), chunk_size): yield self._data[i:i + chunk_size] DASHDRM_OPTIONS = [ "decryption-key", "presentation-delay", "use-subtitles", "ignore-location", "ignore-availability", "availability-grace", "always-play-last-period", "video-codec-preset", "video-timescale", "disable-multi-audio", "language", "use-mp4ff", ] @pluginmatcher( priority=HIGH_PRIORITY, pattern=re.compile(r"dashdrm://(?P\S+)(?:\s(?P.+))?$"), ) @pluginargument( "decryption-key", type="comma_list", help="Decryption key(s) to be passed to ffmpeg." ) @pluginargument( "presentation-delay", help="Override presentation delay value (in seconds). Similar to" " --hls-live-edge." ) @pluginargument( "use-subtitles", action="store_true", help="Experiment with subtitles." ) @pluginargument( "ignore-location", action="store_true", help="Workaround to ignore Location tags that is not compliant resulting" " in wrong segment URLs." ) @pluginargument( "ignore-availability", action="store_true", help="Workaround to ignore segment availability times where server has" " wrong availability segment times (eg times that are in the future)." " Use this with extreme caution as you are ignoring what the server is" " telling you in relation to when a segment is available, thus you can" " overwhelm a server with requests for segments that it already told" " you is not (yet) available." ) @pluginargument( "availability-grace", help="Workaround to delay getting segments even though the segment" " availability time has been reached, as the segments are not" " actually avaiable yet (resulting in 403/404 errors) possibly due to" " mismatched server clock. A negative grace value can also be given," " which can cover cases where the availability times are in the future." ) @pluginargument( "always-play-last-period", action="store_true", help="Always jump to the last period, even when multiple new periods are" " found." ) @pluginargument( "video-codec-preset", help="Use this to specific preset for transcode. This option is meant to be" " paired with --ffmpeg-video-transcode. eg transcoding to libx264 with a" " specific preset (eg ultrafast)" ) @pluginargument( "video-timescale", help="Use this to only display videos with the specifie timescale (eg 50000)" " filtering out any other. This can be used in multi-period mpds to display" " the main content and filter out other content (eg ads) that has incompatible" " timescale which can cause issues with client/players. Warning: Experimental" ) @pluginargument( "disable-multi-audio", action="store_true", help="Restore standard dash plugin function of single audio stream." ) @pluginargument( "language", help="Allow filtering of audio to a specific language code." ) @pluginargument( "use-mp4ff", action="store_true", help="Use mp4ff-decrypt instead of ffmpeg for DRM decryption. " "Requires mp4ff-decrypt to be installed and available in PATH. " ) class MPEGDASHDRM(MPEGDASH): def _get_streams(self): data = self.match.groupdict() url = update_scheme("https://", str(data.get("url", "")), force=False) params = parse_params(data.get("params")) log.debug(f"URL={url}; params={params}") # process and store plugin options before passing streams back for option in DASHDRM_OPTIONS: if option == 'decryption-key': if self.get_option('decryption-key'): self.session.options[option] = self._process_keys() elif option == 'always-play-last-period': if not params.get('period'): params['period'] = -1 else: self.session.options[option] = self.get_option(option) return DASHStreamDRM.parse_manifest(self.session, url, **params) @staticmethod def _extract_kid(obj): if not getattr(obj, 'contentProtections', None): return None for cp in obj.contentProtections: # cp is a streamlink ContentProtection (MPDNode subclass). # cp.node is the underlying lxml _Element. try: raw_attribs = dict(cp.node.attrib) except AttributeError: continue for attr_key, attr_val in raw_attribs.items(): # Strip namespace URI if present: "{urn:...}default_KID" → "default_KID" local_name = attr_key.split('}')[-1] if '}' in attr_key else attr_key if local_name.lower() in ('default_kid',): kid = attr_val.replace('-', '').replace(' ', '').lower() if len(kid) == 32: log.debug('Found KID in ContentProtection attr %s: %s', attr_key, kid) return kid return None @staticmethod def _decode_key_hex(raw_key: str) -> str: key_len = len(raw_key) log.debug('Decryption Key %s has %s digits', raw_key, key_len) if key_len in (21, 22, 23, 24): # Looks like base64 – try to decode it log.debug("Decryption key length is too short to be hex and looks like it might be base64, so we'll try and decode it..") b64_string = raw_key padding = 4 - (len(b64_string) % 4) b64_string = b64_string + ("=" * padding) decoded = base64.urlsafe_b64decode(b64_string).hex() if decoded: raw_key = decoded key_len = len(raw_key) log.debug( 'Decryption Key (post base64 decode) is %s and has %s digits', raw_key, key_len ) if key_len == 32: try: int(raw_key, 16) except ValueError as err: raise FatalPluginError("Expecting 128bit key in 32 hex digits, but the key contains invalid hex.") elif key_len != 32: raise FatalPluginError("Expecting 128bit key in 32 hex digits.") return raw_key.lower() def _process_keys(self): keys = self.get_option('decryption-key') kid_key_map = {} # {kid_norm: key_hex} ordered_keys = [] # positional fallback list for k in keys: parts = k.split(':') if len(parts) == 2: # Format is kid:key kid_raw = parts[0] key_raw = parts[1] kid_norm = kid_raw.replace('-', '').lower() else: # Plain key (no kid) kid_norm = None key_raw = parts[-1] key_hex = self._decode_key_hex(key_raw) if kid_norm: kid_key_map[kid_norm] = key_hex log.debug('KID %s -> Key %s', kid_norm, key_hex) ordered_keys.append(key_hex) log.debug('kid_key_map: %s', kid_key_map) log.debug('ordered_keys: %s', ordered_keys) return {'kid_key_map': kid_key_map, 'ordered_keys': ordered_keys} class FFMPEGMuxerDRM(FFMPEGMuxer): @classmethod def _get_keys(cls, session): keys = [] if session.options.get("use-mp4ff"): return keys key_opt = session.options.get("decryption-key") if key_opt: if isinstance(key_opt, dict): keys = list(key_opt.get("ordered_keys", [])) else: keys = list(key_opt) if len(keys) == 1: keys.extend(keys) log.debug('Decryption Keys %s', keys) return keys def __init__(self, session, *streams, **options): # stream_kids is injected by DASHStreamDRM.open(); pop it before # calling super().__init__() so FFMPEGMuxer doesn't choke on it. super().__init__(session, *streams, **options) keys = self._get_keys(session) key = 0 subtitles = self.session.options.get("use-subtitles") vid_codec_preset = None if session.options.get("video-codec-preset"): vid_codec_preset = self.session.options.get("video-codec-preset") # Rebuild the ffmpeg command inserting -decryption_key before each -i old_cmd = self._cmd.copy() self._cmd = [] while len(old_cmd) > 0: cmd = old_cmd.pop(0) if cmd == "-i": _ = old_cmd.pop(0) if keys: self._cmd.extend(["-decryption_key", keys[key]]) key += 1 if key == len(keys): key = 1 self._cmd.extend(['-thread_queue_size', '4096']) self._cmd.extend([cmd, _]) elif subtitles and cmd == "-c:a": _ = old_cmd.pop(0) self._cmd.extend([cmd, _]) self._cmd.extend(["-c:s", "copy"]) elif vid_codec_preset and cmd == "-c:v": _ = old_cmd.pop(0) self._cmd.extend([cmd, _]) self._cmd.extend(["-preset:v", vid_codec_preset]) else: self._cmd.append(cmd) log.debug("Updated ffmpeg command %s", self._cmd) class DASHStreamWriterDRM(DASHStreamWriter): reader: DASHStreamReaderDRM stream: DASHStreamDRM def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._init_segment_path: str | None = None def __del__(self): self._cleanup_init_segment() def _cleanup_init_segment(self): if self._init_segment_path: with suppress(OSError): os.unlink(self._init_segment_path) self._init_segment_path = None def _get_key_for_stream(self) -> str | None: # Only set if no key was already configured (initial URL / CLI takes priority) key_opt = self.session.options.get("decryption-key") if not key_opt: return None if isinstance(key_opt, dict): kid_key_map = key_opt.get("kid_key_map", {}) ordered_keys = key_opt.get("ordered_keys", []) else: # Parse the raw kid:key strings into the expected dict format kid_key_map = {} ordered_keys = key_opt if not ordered_keys: return None if len(ordered_keys) == 1: return ordered_keys[0] stream_kid = getattr(self.reader, "_drm_kid", None) if stream_kid and kid_key_map: kid_norm = stream_kid.replace('-', '').lower() if kid_norm in kid_key_map: return kid_key_map[kid_norm] mime = getattr(self.reader, "mime_type", "") if mime.startswith("video"): return ordered_keys[0] reader_name = getattr(self.reader, "name", "") try: stream_index = int("".join(filter(str.isdigit, reader_name))) except (ValueError, TypeError): stream_index = 0 audio_key_count = len(ordered_keys) - 1 key_index = 1 + (stream_index % audio_key_count) return ordered_keys[key_index] def _decrypt_segment(self, data: bytes, is_init: bool = False) -> bytes: # Only set if no key was already configured (initial URL / CLI takes priority) key = self._get_key_for_stream() if not key: log.warning("No decryption key available for mp4ff-decrypt, returning raw data") return data # Pour l'init segment : sauvegarder le brut AVANT déchiffrement if is_init: self._cleanup_init_segment() fd, self._init_segment_path = tempfile.mkstemp(suffix="_init_raw.mp4") os.close(fd) with open(self._init_segment_path, "wb") as f: f.write(data) log.debug( "%s raw init segment saved to %s", self.reader.mime_type, self._init_segment_path, ) with tempfile.TemporaryDirectory() as tmpdir: input_path = os.path.join(tmpdir, "segment.mp4") output_path = os.path.join(tmpdir, "segment_dec.mp4") with open(input_path, "wb") as f: f.write(data) if is_init or not self._init_segment_path: cmd = ["mp4ff-decrypt", "-key", key, input_path, output_path] else: cmd = [ "mp4ff-decrypt", "-init", self._init_segment_path, "-key", key, input_path, output_path, ] log.debug("%s mp4ff-decrypt cmd: %s", self.reader.mime_type, cmd) try: result = subprocess.run(cmd, capture_output=True, timeout=30) except FileNotFoundError: raise FatalPluginError( "mp4ff-decrypt not found. Please install mp4ff and make sure " "it is available in your PATH." ) except subprocess.TimeoutExpired: log.error("mp4ff-decrypt timed out on segment, returning raw data") return data if result.returncode != 0: log.error( "mp4ff-decrypt failed (exit %d): %s", result.returncode, result.stderr.decode(errors="replace"), ) return data with open(output_path, "rb") as f: decrypted = f.read() return decrypted def fetch(self, segment: DASHSegment): if self.closed: return real_available_in = (segment.available_at - now()).total_seconds() name = segment.name log.debug( f"{self.reader.mime_type} segment {name}: " f"Available in {real_available_in:.01f}s ({segment.availability})" ) if self.session.options.get("availability-grace"): availability_grace = float(self.session.options.get("availability-grace")) segment.available_at = segment.available_at + timedelta(seconds=availability_grace) log.debug( f"{self.reader.mime_type} segment {name}: " f"Adding {availability_grace} seconds to segment availability" ) if self.session.options.get("ignore-availability"): segment.available_at = fromtimestamp(0) log.debug(f"Ignoring availability timestamps. Now available at {segment.available_at}") response = super().fetch(segment) if not self.session.options.get("use-mp4ff"): return response if response is None: return response raw = b"".join(response.iter_content(8192)) if not raw: return response decrypted = self._decrypt_segment(raw, is_init=bool(segment.init)) return _DecryptedResponse(decrypted) class DASHStreamWorkerDRM(DASHStreamWorker): reader: DASHStreamReaderDRM writer: DASHStreamWriterDRM stream: DASHStreamDRM def iter_segments(self): init = True back_off_factor = 1 new_rep = None queued = False last_segment = None first = True filtered = False video_is_filtered = False repeated = 0 while not self.closed: # find the representation by ID representation = self.mpd.get_representation(self.reader.ident) # check if a new representation is available if not first and not new_rep: new_rep, video_is_filtered = self.check_new_rep() elif first: _, filtered = self.check_new_rep(representation) if filtered: representation = None log.debug("First run. Video filter %s.", filtered) first = False if self.mpd.type == "static": refresh_wait = 5 else: refresh_wait = ( max( self.mpd.minimumUpdatePeriod.total_seconds(), # dont take the whole rep duration as wait time # as some mpd will set a large number. we then # end up staying in the sleeper loop too long # and ffmpeg will timeout min(representation.period.duration.total_seconds(), 5) if representation else 0, ) or 5 ) if new_rep and not queued: # New rep available and no yield so we swap to the new one self.reader.ident = new_rep.ident representation = new_rep new_rep = None log.debug("New period and no yield. Swapping to next period.") if video_is_filtered: log.debug("New period %s marked to be filtered.", self.reader.ident) filtered = True representation = None elif new_rep and queued: # New rep available but we had yield so we dont swap yet. # Set refresh to be very low since we know we actually have # new content in the from of new_rep refresh_wait = 1 log.debug("New period and but we have yield. Not swapping to next period yet.") with self.sleeper(refresh_wait * back_off_factor): if not representation: if filtered: log.debug("Period is marked to be filtered.") if last_segment: repeated += 1 log.debug("Repeating last segment %s times.", repeated) yield last_segment else: log.debug("No last segment to repeat.") self.reload() continue queued = False iter_segments = representation.segments( sequence=self.sequence, init=init, # sync initial timeline generation between audio and video threads timestamp=self.reader.timestamp if init else None, ) for segment in iter_segments: if init and not segment.init: self.sequence = segment.num init = False last_segment = segment repeated = 0 queued |= yield segment # close worker if type is not dynamic (all segments were put into writer queue) if self.mpd.type != "dynamic": self.close() return # Implicit end of stream #if self.check_queue_deadline(queued): # return if not self.reload(): # use min instead of max to limit run-away backoff back_off_factor = min(back_off_factor * 1.3, 10.0) else: back_off_factor = 1 class DASHStreamWorkerDRMVideo(DASHStreamWorkerDRM): reader: DASHStreamReaderDRMVideo writer: DASHStreamWriterDRM stream: DASHStreamDRM def rep_check_filtered(self, first_rep=None): if not first_rep: representation = self.mpd.get_representation(self.reader.ident) else: representation = first_rep if representation.segmentTemplate: current_timescale = representation.segmentTemplate.timescale elif representation.segmentList: current_timescale = representation.segmentList.timescale else: current_timescale = None log.debug("Current rep (%s) timescale: %s", self.reader.mime_type, current_timescale) is_filtered = False if self.session.options.get("video-timescale"): wanted_timescale = int(self.session.options.get("video-timescale")) if current_timescale and (current_timescale != wanted_timescale): is_filtered = True return is_filtered def next_period_available(self): period_id = self.reader.ident[0] current_period_ids = [p.id for p in self.mpd.periods] current_period_idx = current_period_ids.index(period_id) log.debug("Current playing period: %s", current_period_idx + 1) log.debug("Number of periods: %s", len(current_period_ids)) next_period = 0 if len(current_period_ids) > current_period_idx + 1: if self.session.options.get("always-play-last-period"): next_period = current_period_ids[-1] else: next_period = current_period_ids[current_period_idx + 1] audio_count = len(self.stream.audio_representations) if audio_count == 1 and self.stream.audio_representations == [None]: audio_count = 0 if audio_count > 0: for _ in range(audio_count): period_sync_queue.put(next_period) subtitles_count = len(self.stream.subtitles_representations) if subtitles_count == 1 and self.stream.subtitles_representations == [None]: subtitles_count = 0 if subtitles_count > 0: for _ in range(subtitles_count): period_sync_queue.put(next_period) return next_period def check_new_rep(self, first_rep=None): new_video_rep = None is_filtered = False audio_count = len(self.stream.audio_representations) if audio_count == 1 and self.stream.audio_representations == [None]: audio_count = 0 subtitles_count = len(self.stream.subtitles_representations) if subtitles_count == 1 and self.stream.subtitles_representations == [None]: subtitles_count = 0 log.debug("Audio stream count: %s", audio_count) log.debug("Subtitles stream count: %s", subtitles_count) if first_rep: is_filtered = self.rep_check_filtered(first_rep) log.debug("Telling non-video streams first rep and filter") if audio_count > 0: for _ in range(audio_count): rep_sync_queue.put((new_video_rep, is_filtered)) if subtitles_count > 0: for _ in range(subtitles_count): rep_sync_queue.put((new_video_rep, is_filtered)) log.debug("Current rep_sync queue size: %s", rep_sync_queue.qsize()) return (None, is_filtered) log.debug("Checking for new period and representations in video stream") next_period = self.next_period_available() if next_period: # reparse manifest to find the next stream reloaded_streams = DASHStreamDRM.parse_manifest(self.session, self.mpd.url, next_period) p, a, r = self.reader.ident new_video_rep = self.mpd.get_representation((next_period, a, r)) if new_video_rep: log.debug("New video rep found. New ident: %s", new_video_rep.ident) else: log.debug("New period found, but can't find matching video rep, trying to find the stream the old way") reload_stream = reloaded_streams[self.stream.stream_name] new_video_rep = reload_stream.video_representation if new_video_rep: log.debug("New video representation found!") is_filtered = self.rep_check_filtered(new_video_rep) log.debug("Telling non-video streams new rep and filter") if rep_sync_queue.qsize() > 0: log.debug("Current rep_sync queue size is non-zero (%s), waiting 5 seconds", rep_sync_queue.qsize()) self.sleeper(5) if rep_sync_queue.qsize() > 0: log.debug("Current rep_sync queue size is still non-zero (%s).", rep_sync_queue.qsize()) if audio_count > 0: for _ in range(audio_count): rep_sync_queue.put((new_video_rep, is_filtered)) if subtitles_count > 0: for _ in range(subtitles_count): rep_sync_queue.put((new_video_rep, is_filtered)) log.debug("Current rep_sync queue size: %s", rep_sync_queue.qsize()) return (new_video_rep, is_filtered) class DASHStreamWorkerDRMNonVideo(DASHStreamWorkerDRM): reader: DASHStreamReaderDRM writer: DASHStreamWriterDRM stream: DASHStreamDRM def next_period_available(self): period_id = self.reader.ident[0] current_period_ids = [p.id for p in self.mpd.periods] current_period_idx = current_period_ids.index(period_id) log.debug("Current playing period: %s", current_period_idx + 1) log.debug("Number of periods: %s", len(current_period_ids)) next_period = period_sync_queue.get(block=True) return next_period def check_new_rep(self, stream_type=None, first_rep=None): new_stream_rep = None is_filtered = False if first_rep: log.debug("Getting from video streams first rep and filter") log.debug("Current rep_sync queue size just before get: %s", rep_sync_queue.qsize()) _, is_filtered = rep_sync_queue.get(block=True) return (None, is_filtered) log.debug("Checking new reps from video stream") next_period = self.next_period_available() if next_period: log.debug("Got next period %s from video stream", next_period) reloaded_streams = DASHStreamDRM.parse_manifest(self.session, self.mpd.url, next_period) _, a, r = self.reader.ident new_stream_rep = self.mpd.get_representation((next_period, a, r)) if new_stream_rep: log.debug("New %s rep found. New ident: %s", stream_type, new_stream_rep.ident) else: log.debug("New period found, but can't find matching %s rep, trying the old way", stream_type) reload_stream = reloaded_streams[self.stream.stream_name] if stream_type == "audio": audio_num = int(self.name[-1]) new_stream_rep = reload_stream.audio_representations[audio_num] elif stream_type == "sub": new_stream_rep = reload_stream.subtitles_representation if new_stream_rep: log.debug("New %s representation found!", stream_type) log.debug("Checking video rep and filter status from video stream") log.debug("Current rep_sync queue size just before get: %s", rep_sync_queue.qsize()) new_video_rep, is_filtered = rep_sync_queue.get(block=True) log.debug("Video stream sent video rep %s and filter status %s", new_video_rep, is_filtered) log.debug("New video rep: %s, isfiltered: %s", new_video_rep.ident, is_filtered) return (new_stream_rep, is_filtered) return (None, False) class DASHStreamWorkerDRMAudio(DASHStreamWorkerDRMNonVideo): reader: DASHStreamReaderDRMAudio writer: DASHStreamWriterDRM stream: DASHStreamDRM def check_new_rep(self, first_rep=None): return super().check_new_rep(stream_type="audio", first_rep=first_rep) class DASHStreamWorkerDRMSub(DASHStreamWorkerDRMNonVideo): reader: DASHStreamReaderDRMSub writer: DASHStreamWriterDRM stream: DASHStreamDRM def check_new_rep(self, first_rep=None): return super().check_new_rep(stream_type="sub", first_rep=first_rep) class DASHStreamReaderDRM(DASHStreamReader): __worker__ = DASHStreamWorkerDRM __writer__ = DASHStreamWriterDRM worker: DASHStreamWorkerDRM writer: DASHStreamWriterDRM stream: DASHStreamDRM class DASHStreamReaderDRMVideo(DASHStreamReaderDRM): __worker__ = DASHStreamWorkerDRMVideo __writer__ = DASHStreamWriterDRM worker: DASHStreamWorkerDRMVideo writer: DASHStreamWriterDRM stream: DASHStreamDRM class DASHStreamReaderDRMAudio(DASHStreamReaderDRM): __worker__ = DASHStreamWorkerDRMAudio __writer__ = DASHStreamWriterDRM worker: DASHStreamWorkerDRMAudio writer: DASHStreamWriterDRM stream: DASHStreamDRM class DASHStreamReaderDRMSub(DASHStreamReaderDRM): __worker__ = DASHStreamWorkerDRMSub __writer__ = DASHStreamWriterDRM worker: DASHStreamWorkerDRMSub writer: DASHStreamWriterDRM stream: DASHStreamDRM def read(self, size: int) -> bytes: _ = self.buffer.read( size, block=self.writer.is_alive(), timeout=self.timeout, ) log.debug("Subtitle stream segment: %s", _) return _ class DASHStreamDRM(DASHStream): """ Implementation of the MPEG-DASH protocol with DRM support. """ __shortname__ = "dashdrm-mp4ff" def __init__( self, session: Streamlink, mpd: MPD, video_representation: Representation | None = None, audio_representations: List[Representation] | None = None, subtitles_representations: List[Representation] | None = None, duration: float | None = None, **kwargs, ): super().__init__( session, mpd, video_representation, audio_representations[0] if audio_representations[0] else None, duration, **kwargs, ) self.audio_representations = audio_representations self.subtitles_representations = subtitles_representations @staticmethod def parse_mpd(session, manifest: str, mpd_params: Mapping[str, Any]) -> MPD: node = parse_xml(manifest, ignore_ns=True) if session.options.get("ignore-location"): location = node.find('Location') if location is not None: log.warning('Found Location tag: %s', location.text) parent = location.getparent() parent.remove(location) return MPD(node, **mpd_params) @classmethod def parse_manifest( cls, session: Streamlink, url_or_manifest: str, period: int | str = 0, with_video_only: bool = False, with_audio_only: bool = False, **kwargs, ) -> dict[str, DASHStreamDRM]: manifest, mpd_params = cls.fetch_manifest(session, url_or_manifest, **kwargs) # Case 2: decryption_key may appear in the final URL after an API redirect. # fetch_manifest returns res.url (post-redirect), so we check it here. # This covers the case where the initial URL is an API endpoint that # redirects to the real MPD URL with ?decryption_key=... appended. try: mpd = cls.parse_mpd(session, manifest, mpd_params) except Exception as err: raise PluginError(f"Failed to parse MPD manifest: {err}") from err if session.options.get("presentation-delay"): presentation_delay = session.options.get("presentation-delay") mpd.suggestedPresentationDelay = timedelta(seconds=int(presentation_delay)) source = mpd_params.get("url", "MPD manifest") video: list[Representation | None] = [None] if with_audio_only else [] audio: list[Representation | None] = [None] if with_video_only else [] subtitles: list[Representation | None] = [None] if with_audio_only else [] available_periods = [ f"{idx}{f' (id={p.id!r})' if p.id is not None else ''}" for idx, p in enumerate(mpd.periods) ] log.debug(f"Available DASH periods: {', '.join(available_periods)}") try: if isinstance(period, int): period_selection = mpd.periods[period] else: period_selection = mpd.periods_map[period] except LookupError: raise PluginError( f"DASH period {period!r} not found. Select a valid period by index or by id attribute value.", ) from None # Search for suitable video and audio representations for aset in period_selection.adaptationSets: # Extract the KID from the AdaptationSet-level ContentProtection. # The default_KID attribute can sit either on the ContentProtection # element itself (urn:mpeg:dash:mp4protection:2011) or as a # child element. We try several common attribute names. aset_kid = MPEGDASHDRM._extract_kid(aset) if aset_kid: log.debug('AdaptationSet KID: %s', aset_kid) if aset.contentProtections: if not session.options.get("decryption-key"): raise PluginError(f"{source} is protected by DRM but no key given") else: log.debug(f"{source} is protected by DRM") for rep in aset.representations: # KID on the Representation overrides the AdaptationSet KID rep_kid = MPEGDASHDRM._extract_kid(rep) or aset_kid # Attach the KID to the representation so open() can pass it # to FFMPEGMuxerDRM for key matching rep._drm_kid = rep_kid if rep_kid: log.debug('Representation %s KID: %s', rep.id, rep_kid) if rep.contentProtections: if not session.options.get("decryption-key"): raise PluginError(f"{source} is protected by DRM but no key given") else: log.debug(f"{source} is protected by DRM") if rep.mimeType.startswith("video"): video.append(rep) elif rep.mimeType.startswith("audio"): # pragma: no branch audio.append(rep) elif (session.options.get("use-subtitles") and rep.mimeType.startswith("application")): subtitles.append(rep) if not video: video.append(None) if not audio: audio.append(None) if not subtitles: subtitles.append(None) locale = session.localization locale_lang = locale.language lang = None available_languages = set() # if the locale is explicitly set, prefer that language over others for aud in audio: if aud and aud.lang: available_languages.add(aud.lang) with suppress(LookupError): if locale.explicit and aud.lang and Language.get(aud.lang) == locale_lang: lang = aud.lang if not lang: # filter by the first language that appears lang = audio[0].lang if audio[0] else None if session.options.get("language"): lang = session.options.get("language") log.debug( f"Available languages for DASH audio streams: " f"{', '.join(available_languages) or 'NONE'} (using: {lang or 'n/a'})", ) if session.options.get("disable-multi-audio"): if len(available_languages) > 1: audio = [a for a in audio if a and (a.lang is None or a.lang == lang)] # if the language is given by the stream, filter out other languages that do not match #if len(available_languages) > 1: # audio = [a for a in audio if a and (a.lang is None or a.lang == lang)] ret = [] for vid, aud in itertools.product(video, audio): if not vid and not aud: continue # Pass only the selected audio representation, not the full list. # The full list was causing open() to mux ALL audio tracks regardless # of which stream the user selected (e.g. 540p+fr_96k would still # open fr_250k, qaa_96k, qad_96k etc.) if session.options.get("disable-multi-audio"): stream = DASHStreamDRM(session, mpd, vid, [aud], subtitles, **kwargs) else: stream = DASHStreamDRM(session, mpd, vid, audio, subtitles, **kwargs) stream_name = [] if vid: stream_name.append(f"{vid.height or vid.bandwidth_rounded:0.0f}{'p' if vid.height else 'k'}") if aud and len(audio) > 1: stream_name.append(f"a{aud.bandwidth:0.0f}k") ret.append(("+".join(stream_name), stream)) # rename duplicate streams dict_value_list = defaultdict(list) for k, v in ret: dict_value_list[k].append(v) def sortby_bandwidth(dash_stream: DASHStreamDRM) -> float: if dash_stream.video_representation: return dash_stream.video_representation.bandwidth if session.options.get("disable-multi-audio"): if dash_stream.audio_representation: return dash_stream.audio_representation.bandwidth return 0 ret_new = {} for q in dict_value_list: items = dict_value_list[q] with suppress(AttributeError): items = sorted(items, key=sortby_bandwidth, reverse=True) for n in range(len(items)): if n == 0: ret_new[q] = items[n] elif n == 1: ret_new[f"{q}_alt"] = items[n] else: ret_new[f"{q}_alt{n}"] = items[n] # add stream_name to the returned streams so we can find it again for stream_name in ret_new: ret_new[stream_name].stream_name = stream_name return ret_new def open(self): video, audio, audio1 = None, None, None rep_video = self.video_representation rep_audios = self.audio_representations rep_subtitles = self.subtitles_representations timestamp = now() fds = [] # stream_kids tracks the KID for each ffmpeg -i input, in order. # This list is passed to FFMPEGMuxerDRM so it can match kid→key. maps = ["0:v?", "0:a?"] metadata = {} if rep_video: video = DASHStreamReaderDRMVideo(self, rep_video, timestamp, name="video") video._drm_kid = getattr(rep_video, "_drm_kid", None) log.debug(f"Opening DASH reader for: {rep_video.ident!r} - {rep_video.mimeType}") video.open() fds.append(video) next_map = 1 if rep_audios: for i, rep_audio in enumerate(rep_audios): audio = DASHStreamReaderDRMAudio(self, rep_audio, timestamp, name="audio" + str(i)) audio._drm_kid = getattr(rep_audio, "_drm_kid", None) if not audio1: audio1 = audio log.debug(f"Opening DASH reader for: {rep_audio.ident!r} - {rep_audio.mimeType}") audio.open() fds.append(audio) metadata["s:a:{0}".format(i)] = [ "language={0}".format(rep_audio.lang), "title=\"{0}\"".format(rep_audio.lang), ] maps.extend(f"{i}:a" for i in range(next_map, next_map + len(rep_audios))) next_map = len(rep_audios) + 1 # only do subtitles if we have video if rep_subtitles and rep_subtitles[0] and rep_video: for _, rep_subtitle in enumerate(rep_subtitles): subtitle = DASHStreamReaderDRMSub(self, rep_subtitle, timestamp, name="subtitle" + str(_)) subtitle._drm_kid = getattr(rep_subtitle, "_drm_kid", None) log.debug(f"Opening DASH reader for: {rep_subtitle.ident!r} - {rep_subtitle.mimeType}") subtitle.open() fds.append(subtitle) metadata["s:s:{0}".format(_)] = [ "language={0}".format(rep_subtitle.lang), "title=\"{0}\"".format(rep_subtitle.lang), ] maps.extend(f"{_}:s" for _ in range(next_map, next_map + len(rep_subtitles))) if video and audio and FFMPEGMuxerDRM.is_usable(self.session): return FFMPEGMuxerDRM(self.session, *fds, copyts=True, maps=maps, metadata=metadata).open() elif video: return video elif audio: return audio1 __plugin__ = MPEGDASHDRM