mirror of
https://github.com/NohamR/dashdrm-mp4ff.git
synced 2026-07-11 19:10:04 +00:00
08e66f21d7
Add a Streamlink plugin (plugin/dashdrm_mp4ff.py) to support DRM-protected MPEG-DASH streams with optional mp4ff-decrypt integration. Include README with usage and installation notes, an interactive helper (gen.py) that reads vault.json and prints Streamlink commands, a vault.json example, pyproject.toml (declares inquirerpy dependency), .python-version, and .gitignore. This provides the plugin, packaging metadata, and tooling needed to run and test DASH-DRM streams locally.
1076 lines
42 KiB
Python
1076 lines
42 KiB
Python
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<url>\S+)(?:\s(?P<params>.+))?$"),
|
||
)
|
||
@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
|