claudeandClaude Opus 4.8
453ca2f85a
Widevine/DRM research and video download scripts (Apple, CDA, Sky, TVP), CDM tooling, and helper utilities. Excludes downloads/. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WFUhtZEpkGZJjNCGFS9r4Z
1532 lines
58 KiB
Python
Executable File
1532 lines
58 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Download a DRM-protected video from SkyShowtime (Widevine DASH)."""
|
||
|
||
import argparse
|
||
import base64
|
||
import curses
|
||
import getpass
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import string
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import xml.etree.ElementTree as ET
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
import requests
|
||
|
||
HERE = Path(__file__).parent
|
||
TOOLS = HERE / 'tools'
|
||
N_M3U8 = TOOLS / 'N_m3u8DL-RE'
|
||
MP4DEC = TOOLS / 'mp4decrypt'
|
||
SAVE_DIR = HERE / 'downloads'
|
||
TMP_DIR = '/tmp/n_m3u8dl_tmp'
|
||
CREDS_FILE = HERE / '.sky_creds'
|
||
SESSION_FILE = HERE / '.sky_session'
|
||
|
||
class TokenExpiredError(RuntimeError):
|
||
pass
|
||
|
||
|
||
WIDEVINE_SYSTEM_ID = 'edef8ba9-79d6-4ace-a3c8-27dcd51d21ed'
|
||
MPD_NS = '{urn:mpeg:dash:schema:mpd:2011}'
|
||
CENC_NS = '{urn:mpeg:cenc:2013}'
|
||
|
||
APP_ID = 'SKYSHOWTIME-ANDROID-v1'
|
||
SIGNATURE_KEY = base64.b64decode('amZqOXFHZzZhREhhQmJGcEg2d05Fdk42Y0h1SHRaVnBwSFJ2Qmdacw==')
|
||
UA = 'SkyShowtimeAndroid-GLOBAL/4.2.12-121040212'
|
||
|
||
HOST = 'skyshowtime.com'
|
||
OVP = f'https://ovp.{HOST}'
|
||
RANGO = f'https://rango.id.{HOST}'
|
||
ATOM = f'https://atom.{HOST}'
|
||
WEBCLIENTS = f'https://web.clients.{HOST}'
|
||
|
||
SKY_HEADERS = {
|
||
'Accept': 'application/json, text/javascript, */*; q=0.01',
|
||
'User-Agent': UA,
|
||
'x-skyott-activeterritory': 'PL',
|
||
'x-skyott-client-version': '4.3.12',
|
||
'x-skyott-device': 'MOBILE',
|
||
'x-skyott-language': 'pl-PL',
|
||
'x-skyott-platform': 'ANDROID',
|
||
'x-skyott-proposition': 'SKYSHOWTIME',
|
||
'x-skyott-provider': 'SKYSHOWTIME',
|
||
'x-skyott-territory': 'PL',
|
||
}
|
||
|
||
LANG_NAMES = {
|
||
'ar': 'Arabic', 'bg': 'Bulgarian', 'bg-BG': 'Bulgarian',
|
||
'cs': 'Czech', 'cs-CZ': 'Czech',
|
||
'da': 'Danish', 'da-DK': 'Danish',
|
||
'de': 'German', 'de-DE': 'German',
|
||
'el': 'Greek', 'el-GR': 'Greek',
|
||
'en': 'English', 'en-US': 'English (US)', 'en-GB': 'English (UK)',
|
||
'es': 'Spanish', 'es-ES': 'Spanish (Spain)', 'es-419': 'Spanish (Latam)',
|
||
'et': 'Estonian', 'et-EE': 'Estonian',
|
||
'fi': 'Finnish', 'fi-FI': 'Finnish',
|
||
'fr': 'French', 'fr-FR': 'French',
|
||
'hr': 'Croatian', 'hr-HR': 'Croatian',
|
||
'hu': 'Hungarian', 'hu-HU': 'Hungarian',
|
||
'it': 'Italian', 'it-IT': 'Italian',
|
||
'ja': 'Japanese', 'ja-JP': 'Japanese',
|
||
'ko': 'Korean', 'ko-KR': 'Korean',
|
||
'lt': 'Lithuanian', 'lt-LT': 'Lithuanian',
|
||
'lv': 'Latvian', 'lv-LV': 'Latvian',
|
||
'nl': 'Dutch', 'nl-NL': 'Dutch',
|
||
'nb': 'Norwegian', 'nb-NO': 'Norwegian',
|
||
'pl': 'Polish', 'pl-PL': 'Polish',
|
||
'pt': 'Portuguese', 'pt-PT': 'Portuguese (PT)', 'pt-BR': 'Portuguese (BR)',
|
||
'ro': 'Romanian', 'ro-RO': 'Romanian',
|
||
'ru': 'Russian', 'ru-RU': 'Russian',
|
||
'sk': 'Slovak', 'sk-SK': 'Slovak',
|
||
'sl': 'Slovenian', 'sl-SI': 'Slovenian',
|
||
'sr': 'Serbian', 'sr-Latn-ME': 'Montenegrin',
|
||
'sv': 'Swedish', 'sv-SE': 'Swedish',
|
||
'tr': 'Turkish', 'tr-TR': 'Turkish',
|
||
'uk': 'Ukrainian', 'uk-UA': 'Ukrainian',
|
||
'und': 'Undetermined',
|
||
'zh': 'Chinese', 'zh-CN': 'Chinese (Simp.)', 'zh-TW': 'Chinese (Trad.)',
|
||
}
|
||
|
||
|
||
def lang_name(code):
|
||
return LANG_NAMES.get(code, code)
|
||
|
||
|
||
def _find_wvd():
|
||
from pywidevine.device import Device
|
||
for p in sorted((HERE / 'L3').glob('*.wvd')):
|
||
try:
|
||
d = Device.load(str(p))
|
||
if d.system_id and d.system_id != 0:
|
||
return p
|
||
except Exception:
|
||
pass
|
||
return next((HERE / 'L3').glob('*.wvd'), None)
|
||
|
||
WVD = _find_wvd()
|
||
|
||
|
||
def sign(method, url, headers, payload=b''):
|
||
"""Calculate x-sky-signature HMAC-SHA1 header."""
|
||
timestamp = int(time.time())
|
||
path = urlparse(url).path if url.startswith('http') else url
|
||
|
||
text_headers = ''
|
||
for key in sorted(headers.keys()):
|
||
if key.lower().startswith('x-skyott'):
|
||
text_headers += key.lower() + ': ' + headers[key] + '\n'
|
||
headers_md5 = hashlib.md5(text_headers.encode()).hexdigest()
|
||
|
||
if isinstance(payload, str):
|
||
payload = payload.encode('utf-8')
|
||
payload_md5 = hashlib.md5(payload).hexdigest()
|
||
|
||
to_hash = f'{method}\n{path}\n\n{APP_ID}\n1.0\n{headers_md5}\n{timestamp}\n{payload_md5}\n'
|
||
digest = hmac.new(SIGNATURE_KEY, to_hash.encode('utf-8'), hashlib.sha1).digest()
|
||
sig = base64.b64encode(digest).decode('utf-8')
|
||
|
||
return {'x-sky-signature': f'SkyOTT client="{APP_ID}",signature="{sig}",timestamp="{timestamp}",version="1.0"'}
|
||
|
||
|
||
def login(session, username, password):
|
||
url = f'{RANGO}/signin/service/international'
|
||
headers = {k: v for k, v in SKY_HEADERS.items()
|
||
if k not in ('x-skyott-device', 'x-skyott-platform')}
|
||
headers['content-type'] = 'application/x-www-form-urlencoded'
|
||
headers['Accept'] = 'application/vnd.siren+json'
|
||
|
||
r = session.post(url, data={
|
||
'userIdentifier': username,
|
||
'password': password,
|
||
'rememberMe': True,
|
||
'isWeb': True,
|
||
}, headers=headers, timeout=15)
|
||
|
||
try:
|
||
data = r.json()
|
||
except Exception:
|
||
raise RuntimeError(f'Login response is not JSON (status {r.status_code}): {r.text[:200]}')
|
||
|
||
if data.get('properties', {}).get('eventType') != 'success':
|
||
props = data.get('properties', {})
|
||
msg = props.get('userMessage')
|
||
if not msg:
|
||
errors = props.get('errors', {})
|
||
for err in errors.get('categoryErrors', []) + errors.get('fieldErrors', []):
|
||
if err.get('message'):
|
||
msg = f"{err['message']} (code: {err.get('code', '?')})"
|
||
break
|
||
if not msg:
|
||
msg = str(data)[:300]
|
||
raise RuntimeError(f'Login failed: {msg}')
|
||
|
||
cookie_str = '; '.join(f'{k}={v}' for k, v in r.cookies.items())
|
||
return cookie_str
|
||
|
||
|
||
def detect_territory(cookie_str):
|
||
m = re.search(r'(?:hterr|userTerritory|activeTerritory)=([A-Z]{2})', cookie_str)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def get_profile_id(session, cookie_str):
|
||
url = f'{WEBCLIENTS}/bff/personas/v2'
|
||
headers = SKY_HEADERS.copy()
|
||
headers['content-type'] = 'application/json'
|
||
headers['cookie'] = cookie_str
|
||
r = session.post(url, data='', headers=headers, timeout=15)
|
||
data = r.json()
|
||
personas = data.get('personas', [])
|
||
if not personas:
|
||
raise RuntimeError(f'No profiles found: {data}')
|
||
return personas[0]['id']
|
||
|
||
|
||
def get_user_token(session, cookie_str, profile_id, device_id):
|
||
url = f'{OVP}/auth/throttled/tokens'
|
||
post_data = json.dumps({
|
||
'auth': {
|
||
'authScheme': 'MESSO',
|
||
'authIssuer': 'NOWTV',
|
||
'personaId': profile_id,
|
||
'provider': SKY_HEADERS['x-skyott-provider'],
|
||
'providerTerritory': SKY_HEADERS['x-skyott-territory'],
|
||
'proposition': SKY_HEADERS['x-skyott-proposition'],
|
||
},
|
||
'device': {
|
||
'type': 'MOBILE',
|
||
'platform': 'ANDROID',
|
||
'id': device_id,
|
||
'drmDeviceId': 'UNKNOWN',
|
||
}
|
||
})
|
||
headers = {
|
||
'Accept': 'application/vnd.tokens.v1+json',
|
||
'Content-Type': 'application/vnd.tokens.v1+json',
|
||
'cookie': cookie_str,
|
||
}
|
||
headers.update(sign('POST', url, {}, post_data))
|
||
|
||
r = session.post(url, data=post_data, headers=headers, timeout=15)
|
||
data = r.json()
|
||
if 'userToken' not in data:
|
||
raise RuntimeError(f'Failed to get user token: {data}')
|
||
return data['userToken']
|
||
|
||
|
||
def get_video_info(session, uuid):
|
||
url = (f'{ATOM}/adapter-calypso/v3/query/nodes/uuid/{uuid}'
|
||
f'?exclude=expired%2Cfuture%2Cshortform')
|
||
r = session.get(url, headers=SKY_HEADERS, timeout=15)
|
||
data = r.json()
|
||
if not data:
|
||
raise RuntimeError(f'No video info returned for UUID {uuid}')
|
||
|
||
att = data[0]['attributes']
|
||
title = att.get('title', uuid)
|
||
provider_variant_id = att.get('providerVariantId')
|
||
content_id = None
|
||
|
||
for fmt in ('HD', 'SD'):
|
||
if fmt in att.get('formats', {}):
|
||
content_id = att['formats'][fmt].get('contentId')
|
||
if content_id:
|
||
break
|
||
|
||
if not content_id:
|
||
raise RuntimeError('No contentId found in video info')
|
||
|
||
return title, content_id, provider_variant_id
|
||
|
||
|
||
def get_playback_info(session, user_token, content_id, provider_variant_id):
|
||
url = f'{OVP}/video/playouts/vod'
|
||
headers = SKY_HEADERS.copy()
|
||
headers['Accept'] = 'application/vnd.playvod.v1+json'
|
||
headers['Content-Type'] = 'application/vnd.playvod.v1+json'
|
||
headers['x-skyott-usertoken'] = user_token
|
||
|
||
post_data = json.dumps({
|
||
'device': {
|
||
'capabilities': [
|
||
{'protection': 'WIDEVINE', 'container': 'ISOBMFF',
|
||
'transport': 'DASH', 'acodec': 'AAC', 'vcodec': 'H264'},
|
||
{'protection': 'NONE', 'container': 'ISOBMFF',
|
||
'transport': 'DASH', 'acodec': 'AAC', 'vcodec': 'H264'},
|
||
],
|
||
'maxVideoFormat': 'HD',
|
||
'model': 'Pixel',
|
||
'hdcpEnabled': False,
|
||
'supportedColourSpaces': ['SDR'],
|
||
},
|
||
'client': {'thirdParties': ['FREEWHEEL']},
|
||
'parentalControlPin': 'null',
|
||
'personaParentalControlRating': '19',
|
||
'contentId': content_id,
|
||
'providerVariantId': provider_variant_id,
|
||
})
|
||
|
||
headers.update(sign('POST', url, headers, post_data))
|
||
r = session.post(url, data=post_data, headers=headers, timeout=15)
|
||
data = r.json()
|
||
|
||
if r.status_code in (401, 403) or (
|
||
'asset' not in data and
|
||
data.get('status', 0) in (401, 403)):
|
||
raise TokenExpiredError(f'User token rejected (HTTP {r.status_code}): {json.dumps(data)[:200]}')
|
||
|
||
if 'asset' not in data:
|
||
raise RuntimeError(f'No asset in playback response: {json.dumps(data)[:400]}')
|
||
|
||
manifest_url = data['asset']['endpoints'][0]['url']
|
||
manifest_url += '&audio=all&subtitle=all&forcedNarrative=true&trickplay=true'
|
||
license_url = data['protection']['licenceAcquisitionUrl']
|
||
|
||
return manifest_url, license_url
|
||
|
||
|
||
def fetch_mpd(mpd_url, session):
|
||
"""Fetch MPD and return (text, parsed root element)."""
|
||
r = session.get(mpd_url, headers={'User-Agent': UA}, timeout=15)
|
||
r.raise_for_status()
|
||
return r.text, ET.fromstring(r.text)
|
||
|
||
|
||
def get_pssh_from_root(root):
|
||
"""Extract Widevine PSSH base64 from a parsed MPD root element."""
|
||
for cp in root.iter(f'{MPD_NS}ContentProtection'):
|
||
if 'edef8ba9' in cp.get('schemeIdUri', '').lower():
|
||
pssh_el = cp.find(f'{CENC_NS}pssh')
|
||
if pssh_el is not None and pssh_el.text:
|
||
return pssh_el.text.strip()
|
||
raise RuntimeError('No Widevine PSSH found in MPD.')
|
||
|
||
|
||
def _stream_path(as_el, codec, ch_value, lang):
|
||
"""Extract or derive the stream path used in N_m3u8DL-RE's stream name.
|
||
|
||
N_m3u8DL-RE names streams as "r{id}/{path}" (e.g. r21/audio/surround/en-US).
|
||
We derive the path so we can use the name= filter to select individual tracks.
|
||
"""
|
||
# 1. BaseURL within AdaptationSet (most reliable)
|
||
bu = as_el.find(f'{MPD_NS}BaseURL')
|
||
if bu is not None and bu.text:
|
||
text = bu.text.strip().rstrip('/')
|
||
if '://' not in text and text:
|
||
return text
|
||
|
||
# 2. SegmentTemplate initialization / media path
|
||
st = as_el.find(f'{MPD_NS}SegmentTemplate')
|
||
if st is not None:
|
||
for attr in ('initialization', 'media'):
|
||
raw = st.get(attr, '').lstrip('/')
|
||
if raw:
|
||
parts = raw.split('/')
|
||
clean = []
|
||
for p in parts:
|
||
if '$' in p or '.' in p:
|
||
break
|
||
clean.append(p)
|
||
if clean:
|
||
return '/'.join(clean)
|
||
|
||
# 3. Construct from codec + channel + lang (SkyShowtime CMAF convention)
|
||
if 'ec-3' in codec.lower():
|
||
fmt = 'surround' if ch_value.upper().startswith('F') else 'stereo-eac3'
|
||
elif 'mp4a' in codec.lower():
|
||
fmt = 'stereo'
|
||
else:
|
||
fmt = 'audio'
|
||
return f'audio/{fmt}/{lang}'
|
||
|
||
|
||
def parse_mpd_tracks(root):
|
||
"""Parse video, audio AdaptationSets and subtitle tracks from MPD.
|
||
|
||
Returns:
|
||
video_streams: list of dicts per Representation (sorted best-first):
|
||
{width, height, kbps, codec}
|
||
audio_tracks: list of dicts per AdaptationSet (individual format variant):
|
||
{lang, lang_disp, fmt_type, codec_disp, kbps, ch_value, name_pattern}
|
||
sub_langs: dict {lang_code: {'forced_only': bool}}
|
||
sub_tracks: list of (lang, is_forced) in MPD order — index N here maps to
|
||
ffmpeg stream 0:s:N when downloading from the same MPD
|
||
"""
|
||
video_streams = []
|
||
audio_tracks = []
|
||
sub_langs = {}
|
||
sub_tracks = [] # (lang, is_forced) in MPD AdaptationSet order
|
||
|
||
for as_el in root.iter(f'{MPD_NS}AdaptationSet'):
|
||
mime = as_el.get('mimeType', '')
|
||
ctype = as_el.get('contentType', '')
|
||
lang = as_el.get('lang', 'und')
|
||
|
||
if ctype == 'video' or mime.startswith('video/'):
|
||
# Skip thumbnail / trickplay tracks
|
||
if any(r.get('value', '') in ('thumbnail', 'trick', 'trickplay')
|
||
for r in as_el.iter(f'{MPD_NS}Role')):
|
||
continue
|
||
as_codec = as_el.get('codecs', '')
|
||
for rep in as_el.iter(f'{MPD_NS}Representation'):
|
||
w = int(rep.get('width', 0))
|
||
h = int(rep.get('height', 0))
|
||
bw = int(rep.get('bandwidth', 0)) // 1000
|
||
codec = rep.get('codecs', '') or as_codec
|
||
if w > 0 and h > 0 and bw >= 200:
|
||
video_streams.append({'width': w, 'height': h, 'kbps': bw, 'codec': codec})
|
||
|
||
elif ctype == 'audio' or mime.startswith('audio/'):
|
||
# Best (highest bandwidth) representation gives codec + kbps
|
||
max_bw = 0
|
||
codec = as_el.get('codecs', '')
|
||
for rep in as_el.iter(f'{MPD_NS}Representation'):
|
||
bw = int(rep.get('bandwidth', 0))
|
||
if bw > max_bw:
|
||
max_bw = bw
|
||
c = rep.get('codecs', '')
|
||
if c:
|
||
codec = c
|
||
kbps = max_bw // 1000
|
||
|
||
# Channel configuration value
|
||
ch_value = ''
|
||
for acc in as_el.iter(f'{MPD_NS}AudioChannelConfiguration'):
|
||
ch_value = acc.get('value', '')
|
||
if ch_value:
|
||
break
|
||
|
||
# Human-readable format type
|
||
if 'ec-3' in codec.lower():
|
||
codec_disp = 'EC-3'
|
||
fmt_type = 'Surround 5.1' if ch_value.upper().startswith('F') else 'Stereo'
|
||
elif 'mp4a' in codec.lower():
|
||
codec_disp = 'AAC'
|
||
try:
|
||
fmt_type = 'Stereo' if int(ch_value) <= 2 else f'Surround {ch_value}ch'
|
||
except (ValueError, TypeError):
|
||
fmt_type = 'Stereo'
|
||
else:
|
||
codec_disp = codec
|
||
fmt_type = ''
|
||
|
||
# Path used in segment URLs (e.g. "audio/surround/en-US")
|
||
name_pattern = _stream_path(as_el, codec, ch_value, lang)
|
||
|
||
audio_tracks.append({
|
||
'lang': lang,
|
||
'lang_disp': lang_name(lang),
|
||
'fmt_type': fmt_type,
|
||
'codec_disp': codec_disp,
|
||
'kbps': kbps,
|
||
'ch_value': ch_value,
|
||
'name_pattern': name_pattern,
|
||
})
|
||
|
||
elif ctype == 'text' or mime.startswith('text/'):
|
||
roles = list(as_el.iter(f'{MPD_NS}Role'))
|
||
is_forced = bool(roles) and all(
|
||
r.get('value', '') in ('forced', 'forcedSubtitle') for r in roles
|
||
)
|
||
sub_tracks.append((lang, is_forced))
|
||
|
||
if lang not in sub_langs:
|
||
sub_langs[lang] = {'forced_only': True}
|
||
if not roles or any(r.get('value', '') not in ('forced', 'forcedSubtitle') for r in roles):
|
||
sub_langs[lang]['forced_only'] = False
|
||
|
||
# Sort video best-first (highest resolution, then highest bitrate)
|
||
video_streams.sort(key=lambda x: (-x['width'], -x['height'], -x['kbps']))
|
||
|
||
return video_streams, audio_tracks, sub_langs, sub_tracks
|
||
|
||
|
||
def checkbox_picker(title, items, default_all=True, single_select=False, default_idx=None):
|
||
"""Interactive curses checkbox / radio-button picker.
|
||
|
||
single_select: if True, Space selects one item and deselects all others.
|
||
default_idx: pre-select this index only (overrides default_all).
|
||
Returns list of selected indices.
|
||
"""
|
||
if not items:
|
||
return []
|
||
|
||
if default_idx is not None:
|
||
selected = [i == default_idx for i in range(len(items))]
|
||
else:
|
||
selected = [default_all] * len(items)
|
||
|
||
state = {'cursor': 0, 'offset': 0}
|
||
|
||
def _run(stdscr):
|
||
curses.curs_set(0)
|
||
try:
|
||
curses.start_color()
|
||
curses.use_default_colors()
|
||
curses.init_pair(1, curses.COLOR_BLACK, curses.COLOR_CYAN)
|
||
except Exception:
|
||
pass
|
||
|
||
while True:
|
||
stdscr.erase()
|
||
h, w = stdscr.getmaxyx()
|
||
list_rows = max(1, h - 5)
|
||
cursor = state['cursor']
|
||
offset = state['offset']
|
||
|
||
if cursor < offset:
|
||
offset = cursor
|
||
elif cursor >= offset + list_rows:
|
||
offset = cursor - list_rows + 1
|
||
state['offset'] = offset
|
||
|
||
try:
|
||
stdscr.addstr(0, 0, title[:w - 1], curses.A_BOLD)
|
||
stdscr.addstr(1, 0, ('─' * min(len(title), w - 1))[:w - 1])
|
||
if single_select:
|
||
hint = '↑↓ move SPACE select ENTER=confirm'
|
||
else:
|
||
hint = '↑↓ move SPACE toggle a=all n=none ENTER=confirm'
|
||
stdscr.addstr(2, 0, hint[:w - 1])
|
||
except curses.error:
|
||
pass
|
||
|
||
for i in range(min(list_rows, len(items))):
|
||
idx = i + offset
|
||
if idx >= len(items):
|
||
break
|
||
y = i + 4
|
||
mark = ('(◉)' if selected[idx] else '( )') if single_select else ('[✓]' if selected[idx] else '[ ]')
|
||
line = f' {mark} {items[idx]}'[:w - 1]
|
||
try:
|
||
attr = curses.color_pair(1) if idx == cursor else 0
|
||
stdscr.addstr(y, 0, line, attr)
|
||
except curses.error:
|
||
pass
|
||
|
||
if not single_select:
|
||
sel_count = sum(selected)
|
||
try:
|
||
stdscr.addstr(h - 1, 0,
|
||
f' {sel_count} of {len(items)} selected'[:w - 1], curses.A_DIM)
|
||
except curses.error:
|
||
pass
|
||
|
||
stdscr.refresh()
|
||
key = stdscr.getch()
|
||
|
||
if key in (curses.KEY_UP, ord('k')):
|
||
state['cursor'] = max(0, cursor - 1)
|
||
elif key in (curses.KEY_DOWN, ord('j')):
|
||
state['cursor'] = min(len(items) - 1, cursor + 1)
|
||
elif key == ord(' '):
|
||
if single_select:
|
||
for j in range(len(selected)):
|
||
selected[j] = (j == cursor)
|
||
else:
|
||
selected[cursor] = not selected[cursor]
|
||
elif not single_select and key == ord('a'):
|
||
for j in range(len(selected)):
|
||
selected[j] = True
|
||
elif not single_select and key == ord('n'):
|
||
for j in range(len(selected)):
|
||
selected[j] = False
|
||
elif key in (curses.KEY_ENTER, 10, 13):
|
||
break
|
||
|
||
try:
|
||
curses.wrapper(_run)
|
||
except Exception as e:
|
||
print(f'[warning] Interactive picker failed ({e}), using default.')
|
||
if single_select:
|
||
fallback = default_idx if default_idx is not None else 0
|
||
return [fallback] if fallback < len(items) else [0]
|
||
return list(range(len(items)))
|
||
|
||
return [i for i, s in enumerate(selected) if s]
|
||
|
||
|
||
def _audio_select_args(lang_pattern, sel_fmts, all_fmts, audio_tracks):
|
||
"""Build the most precise single --select-audio filter achievable.
|
||
|
||
N_m3u8DL-RE only accepts --select-audio once and ch= is ignored for audio.
|
||
Confirmed working attributes: lang=, codecs=, bwMin=, bwMax= (all in Kbps).
|
||
|
||
Strategy per selected format combination:
|
||
• All formats → lang=X:for=all
|
||
• AAC only → lang=X:codecs=mp4a:for=all
|
||
• All EC-3 → lang=X:codecs=ec-3:for=all
|
||
• High-bw EC-3 only → lang=X:codecs=ec-3:bwMin=T:for=all (surround tier)
|
||
• Low-bw EC-3 only → lang=X:codecs=ec-3:bwMax=T:for=all (stereo-EC-3 tier)
|
||
• Mixed codecs → lang=X:for=all (can't express cross-codec OR in one filter)
|
||
"""
|
||
if not sel_fmts:
|
||
return ['--drop-audio', 'all']
|
||
if len(sel_fmts) == len(all_fmts):
|
||
return ['--select-audio', f'lang={lang_pattern}:for=all']
|
||
|
||
sel_ec3 = [f for f in sel_fmts if f.get('codec_disp') == 'EC-3']
|
||
sel_aac = [f for f in sel_fmts if f.get('codec_disp') == 'AAC']
|
||
all_ec3 = [f for f in all_fmts if f.get('codec_disp') == 'EC-3']
|
||
|
||
# AAC only
|
||
if sel_aac and not sel_ec3:
|
||
return ['--select-audio', f'lang={lang_pattern}:codecs=mp4a:for=all']
|
||
|
||
# EC-3 only (some or all EC-3 tiers)
|
||
if sel_ec3 and not sel_aac:
|
||
if len(sel_ec3) == len(all_ec3):
|
||
return ['--select-audio', f'lang={lang_pattern}:codecs=ec-3:for=all']
|
||
|
||
# Multiple EC-3 tiers exist — distinguish by bitrate threshold
|
||
ec3_bws = sorted(set(f['kbps'] for f in all_ec3), reverse=True) # [388, 132]
|
||
sel_bws = {f['kbps'] for f in sel_ec3}
|
||
if len(ec3_bws) >= 2:
|
||
threshold = (ec3_bws[0] + ec3_bws[-1]) // 2 # e.g. 260 Kbps
|
||
hi_selected = any(bw >= ec3_bws[0] - 20 for bw in sel_bws)
|
||
lo_selected = any(bw <= ec3_bws[-1] + 20 for bw in sel_bws)
|
||
if hi_selected and not lo_selected:
|
||
return ['--select-audio', f'lang={lang_pattern}:codecs=ec-3:bwMin={threshold}:for=all']
|
||
if lo_selected and not hi_selected:
|
||
return ['--select-audio', f'lang={lang_pattern}:codecs=ec-3:bwMax={threshold}:for=all']
|
||
|
||
return ['--select-audio', f'lang={lang_pattern}:codecs=ec-3:for=all']
|
||
|
||
# Mixed EC-3 + AAC: can't express the cross-codec OR, fall back to all langs
|
||
return ['--select-audio', f'lang={lang_pattern}:for=all']
|
||
|
||
|
||
def select_tracks_interactive(audio_tracks, sub_langs):
|
||
"""Show pickers for audio languages + format types, and subtitle languages.
|
||
|
||
Audio selection is two steps:
|
||
1. Which languages to include?
|
||
2. Which format types to include?
|
||
|
||
Returns (audio_args, sel_sub_langs) where sel_sub_langs is a set of
|
||
selected language codes (empty set = no subtitles).
|
||
"""
|
||
# ── Build unique language list (preserving MPD order) ──────────────────
|
||
seen_langs: set = set()
|
||
lang_list: list = []
|
||
for t in audio_tracks:
|
||
if t['lang'] not in seen_langs:
|
||
seen_langs.add(t['lang'])
|
||
lang_list.append(t['lang'])
|
||
|
||
lang_items = [f'{l:<9} {lang_name(l)}' for l in lang_list]
|
||
|
||
# ── Build unique format-type list (preserving MPD order) ───────────────
|
||
seen_fmt: set = set()
|
||
fmt_list: list = []
|
||
for t in audio_tracks:
|
||
key = (t['fmt_type'], t['codec_disp'], t['ch_value'])
|
||
if key not in seen_fmt:
|
||
seen_fmt.add(key)
|
||
fmt_list.append({
|
||
'fmt_type': t['fmt_type'],
|
||
'codec_disp': t['codec_disp'],
|
||
'ch_value': t['ch_value'],
|
||
'kbps': t['kbps'],
|
||
})
|
||
|
||
fmt_items = []
|
||
for f in fmt_list:
|
||
kbps_str = f'~{f["kbps"]} Kbps' if f['kbps'] else ''
|
||
fmt_items.append(
|
||
f'{f["fmt_type"]:<16} {f["codec_disp"]:<8}' + (f' {kbps_str}' if kbps_str else '')
|
||
)
|
||
|
||
# ── Audio picker ────────────────────────────────────────────────────────
|
||
print()
|
||
audio_args = ['--drop-audio', 'all']
|
||
|
||
if lang_items and fmt_items:
|
||
sel_lang_idx = checkbox_picker('Select audio languages', lang_items, default_all=False)
|
||
sel_fmt_idx = checkbox_picker('Select audio format types', fmt_items, default_all=False)
|
||
|
||
sel_langs = [lang_list[i] for i in sel_lang_idx]
|
||
sel_fmts = [fmt_list[i] for i in sel_fmt_idx]
|
||
|
||
if sel_langs and sel_fmts:
|
||
lang_pattern = '|'.join(sel_langs)
|
||
audio_args = _audio_select_args(lang_pattern, sel_fmts, fmt_list, audio_tracks)
|
||
fmt_labels = ', '.join(f'{f["fmt_type"]} ({f["codec_disp"]})' for f in sel_fmts)
|
||
print(f'Audio: {", ".join(sel_langs)} — {fmt_labels}')
|
||
elif sel_langs:
|
||
lang_pattern = '|'.join(sel_langs)
|
||
audio_args = ['--select-audio', f'lang={lang_pattern}:for=all']
|
||
print(f'Audio: {", ".join(sel_langs)} — all formats')
|
||
else:
|
||
print('Audio: none')
|
||
|
||
# ── Subtitle picker ─────────────────────────────────────────────────────
|
||
sub_keys = list(sub_langs.keys())
|
||
sub_items = []
|
||
for lang in sub_keys:
|
||
note = ' (forced only)' if sub_langs[lang]['forced_only'] else ''
|
||
sub_items.append(f'{lang:<9} {lang_name(lang)}{note}')
|
||
|
||
print()
|
||
sel_sub_langs: set = set()
|
||
if sub_items:
|
||
sel_idx = checkbox_picker('Select subtitle tracks', sub_items, default_all=False)
|
||
sel_subs = [sub_keys[i] for i in sel_idx]
|
||
if sel_subs:
|
||
sel_sub_langs = set(sel_subs)
|
||
print(f'Subtitles: {", ".join(sel_subs)}')
|
||
else:
|
||
print('Subtitles: none')
|
||
|
||
return audio_args, sel_sub_langs
|
||
|
||
|
||
def select_video_interactive(video_streams):
|
||
"""Show single-select picker for video quality. Returns N_m3u8DL-RE --select-video args."""
|
||
if not video_streams:
|
||
return ['--select-video', 'best']
|
||
|
||
# Count streams per resolution to detect duplicates
|
||
from collections import Counter
|
||
res_count = Counter(f'{v["width"]}x{v["height"]}' for v in video_streams)
|
||
|
||
items = []
|
||
for vs in video_streams:
|
||
codec = vs.get('codec', '')
|
||
if 'hvc' in codec.lower() or 'hev' in codec.lower() or 'dvh' in codec.lower():
|
||
codec_name = 'H.265'
|
||
elif 'avc' in codec.lower():
|
||
codec_name = 'H.264'
|
||
else:
|
||
codec_name = codec.split('.')[0] if codec else '?'
|
||
items.append(f'{vs["width"]}×{vs["height"]:<9} {vs["kbps"]:>5} Kbps {codec_name}')
|
||
|
||
print()
|
||
sel_idx = checkbox_picker('Select video quality', items, single_select=True, default_idx=0)
|
||
|
||
if not sel_idx:
|
||
return ['--select-video', 'best']
|
||
|
||
chosen = video_streams[sel_idx[0]]
|
||
w, h, kbps = chosen['width'], chosen['height'], chosen['kbps']
|
||
res_str = f'{w}x{h}'
|
||
|
||
if res_count[res_str] > 1:
|
||
# Duplicate resolution: use bitrate range (±20 %) to single out this tier.
|
||
# N_m3u8DL-RE bwMin/bwMax are in Kbps.
|
||
bw_min = int(kbps * 0.8)
|
||
filter_str = f'res={res_str}:bwMin={bw_min}:for=best'
|
||
else:
|
||
filter_str = f'res={res_str}:for=best'
|
||
|
||
print(f'Video: {w}×{h} {kbps} Kbps')
|
||
return ['--select-video', filter_str]
|
||
|
||
|
||
def langs_to_select_args(select_flag, drop_flag, langs):
|
||
"""Build N_m3u8DL-RE stream selection args from a list of language codes."""
|
||
if not langs:
|
||
return [drop_flag, 'all']
|
||
pattern = '|'.join(langs)
|
||
return [select_flag, f'lang={pattern}:for=all']
|
||
|
||
|
||
def cli_spec_to_select_args(select_flag, drop_flag, spec):
|
||
"""Build N_m3u8DL-RE stream selection args from a CLI spec string.
|
||
|
||
spec='all' → select all
|
||
spec='none' → drop all
|
||
spec='pl-PL,en-US' → select by language
|
||
"""
|
||
spec = spec.strip()
|
||
if spec == 'all':
|
||
return [select_flag, 'all']
|
||
if spec == 'none':
|
||
return [drop_flag, 'all']
|
||
langs = [s.strip() for s in spec.split(',') if s.strip()]
|
||
return langs_to_select_args(select_flag, drop_flag, langs)
|
||
|
||
|
||
def fetch_content_key(pssh_b64, license_url):
|
||
from pywidevine.cdm import Cdm
|
||
from pywidevine.device import Device
|
||
from pywidevine.pssh import PSSH
|
||
import uuid as _uuid
|
||
|
||
device = Device.load(str(WVD))
|
||
cdm = Cdm.from_device(device)
|
||
session_id = cdm.open()
|
||
challenge = cdm.get_license_challenge(session_id, PSSH(pssh_b64))
|
||
|
||
path = '/' + license_url.split('://', 1)[1].split('/', 1)[1]
|
||
req_headers = {'User-Agent': UA, 'Accept': '*/*'}
|
||
req_headers.update(sign('POST', path, {}, challenge))
|
||
|
||
r = requests.post(license_url, data=challenge, headers=req_headers, timeout=15)
|
||
if not r.ok:
|
||
raise RuntimeError(f'License server {r.status_code}: {r.text[:300]}')
|
||
|
||
cdm.parse_license(session_id, r.content)
|
||
|
||
keys = {}
|
||
for key in cdm.get_keys(session_id):
|
||
if str(key.type) not in ('SIGNING', 'OPERATOR_SESSION'):
|
||
kid = key.kid.hex if isinstance(key.kid, _uuid.UUID) else key.kid.hex()
|
||
keys[kid] = key.key.hex()
|
||
cdm.close(session_id)
|
||
return keys
|
||
|
||
|
||
def extract_uuid(url):
|
||
m = re.search(r'([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})',
|
||
url, re.IGNORECASE)
|
||
if not m:
|
||
raise ValueError(f'Cannot extract UUID from URL: {url}')
|
||
return m.group(1)
|
||
|
||
|
||
def safe_filename(title):
|
||
title = re.sub(r'[^\w\s-]', '', title)
|
||
title = re.sub(r'[\s-]+', '_', title).strip('_')
|
||
return title
|
||
|
||
|
||
_ISO639_2_TO_1 = {
|
||
'pol': 'pl', 'eng': 'en', 'deu': 'de', 'ger': 'de', 'fra': 'fr', 'fre': 'fr',
|
||
'spa': 'es', 'ita': 'it', 'por': 'pt', 'rus': 'ru', 'ces': 'cs', 'cze': 'cs',
|
||
'slk': 'sk', 'slv': 'sl', 'hun': 'hu', 'ron': 'ro', 'rum': 'ro', 'bul': 'bg',
|
||
'hrv': 'hr', 'srp': 'sr', 'nob': 'nb', 'nor': 'nb', 'dan': 'da', 'swe': 'sv',
|
||
'fin': 'fi', 'nld': 'nl', 'dut': 'nl', 'tur': 'tr', 'ukr': 'uk', 'ara': 'ar',
|
||
'zho': 'zh', 'chi': 'zh', 'jpn': 'ja', 'kor': 'ko', 'ell': 'el', 'gre': 'el',
|
||
'lat': 'la', 'und': 'und',
|
||
}
|
||
|
||
|
||
def _lang_tag_to_iso639_1(tag):
|
||
"""Return the ISO 639-1 two-letter code for a language tag.
|
||
Handles ISO 639-2 three-letter codes (pol→pl), BCP-47 (pl-PL→pl),
|
||
and bare ISO 639-1 (pl→pl).
|
||
"""
|
||
t = tag.strip().lower()
|
||
if t in _ISO639_2_TO_1:
|
||
return _ISO639_2_TO_1[t]
|
||
return t.split('-')[0].split('_')[0]
|
||
|
||
|
||
def _parse_vtt_ts(ts):
|
||
"""Convert a VTT timestamp (HH:MM:SS.mmm or MM:SS.mmm) to SRT format (HH:MM:SS,mmm)."""
|
||
ts = ts.strip()
|
||
parts = ts.split(':')
|
||
if len(parts) == 3:
|
||
h, m, s = parts
|
||
elif len(parts) == 2:
|
||
h, m, s = '0', parts[0], parts[1]
|
||
else:
|
||
return None
|
||
return f'{int(h):02d}:{m}:{s.replace(".", ",")}'
|
||
|
||
|
||
def _vtt_to_srt(vtt_text):
|
||
"""Convert merged WebVTT text to SRT.
|
||
|
||
Handles STYLE/NOTE/REGION blocks, X-TIMESTAMP-MAP headers, and VTT-specific
|
||
inline tags (<c>, <v>, <ruby>) — none of which ffmpeg's WebVTT parser
|
||
handles reliably when all segments are merged into one file.
|
||
"""
|
||
lines = vtt_text.replace('\r\n', '\n').replace('\r', '\n').split('\n')
|
||
n = len(lines)
|
||
i = 0
|
||
|
||
# Skip WEBVTT header line and any header-metadata lines (X-TIMESTAMP-MAP etc.)
|
||
if i < n and lines[i].startswith('WEBVTT'):
|
||
i += 1
|
||
while i < n and lines[i].strip():
|
||
i += 1 # skip header continuation until first blank
|
||
|
||
cues = []
|
||
|
||
while i < n:
|
||
line = lines[i].strip()
|
||
|
||
if not line:
|
||
i += 1
|
||
continue
|
||
|
||
# Skip block types that are not cues
|
||
if line in ('STYLE', 'REGION') or line.startswith('NOTE'):
|
||
i += 1
|
||
while i < n and lines[i].strip():
|
||
i += 1
|
||
continue
|
||
|
||
# Determine if this line is a timestamp or a cue identifier
|
||
if '-->' in line:
|
||
ts_line = line
|
||
else:
|
||
# Treat as cue ID; the next non-empty line should be the timestamp
|
||
i += 1
|
||
while i < n and not lines[i].strip():
|
||
i += 1
|
||
if i >= n or '-->' not in lines[i]:
|
||
continue
|
||
ts_line = lines[i].strip()
|
||
|
||
# Parse "start --> end [settings]"
|
||
arrow = ts_line.index('-->')
|
||
start_srt = _parse_vtt_ts(ts_line[:arrow])
|
||
end_srt = _parse_vtt_ts(ts_line[arrow + 3:].split()[0])
|
||
if not start_srt or not end_srt:
|
||
i += 1
|
||
continue
|
||
|
||
# Read cue text lines until blank line
|
||
i += 1
|
||
text_lines = []
|
||
while i < n and lines[i].strip():
|
||
cue_line = lines[i].strip()
|
||
# Strip WebVTT-specific tags: <c.class>, </c>, <v Name>, <ruby>, <rt>
|
||
cue_line = re.sub(r'</?c(\.[^>]*)?>', '', cue_line)
|
||
cue_line = re.sub(r'<v [^>]+>', '', cue_line)
|
||
cue_line = re.sub(r'</?r(?:uby|[bt])[^>]*>', '', cue_line)
|
||
# Strip timestamp tags <HH:MM:SS.mmm>
|
||
cue_line = re.sub(r'<\d{2}:\d{2}:\d{2}\.\d{3}>', '', cue_line)
|
||
cue_line = cue_line.strip()
|
||
if cue_line:
|
||
text_lines.append(cue_line)
|
||
i += 1
|
||
|
||
if text_lines:
|
||
cues.append((start_srt, end_srt, text_lines))
|
||
|
||
parts = [
|
||
f'{num}\n{s} --> {e}\n' + '\n'.join(t)
|
||
for num, (s, e, t) in enumerate(cues, 1)
|
||
]
|
||
return '\n\n'.join(parts) + ('\n' if parts else '')
|
||
|
||
|
||
def _dash_segment_url(tmpl, rep_id, number=None, time=None):
|
||
"""Expand a DASH SegmentTemplate media URL."""
|
||
def _replace(m):
|
||
name = m.group(1)
|
||
fmt = m.group(2) # e.g. '%06d' captured by (%\d+[diouxX])?, or None
|
||
if name == 'RepresentationID':
|
||
return rep_id
|
||
if name == 'Number' and number is not None:
|
||
return (fmt % number) if fmt else str(number)
|
||
if name == 'Time' and time is not None:
|
||
return (fmt % time) if fmt else str(time)
|
||
return m.group(0)
|
||
return re.sub(r'\$(\w+)(%\d+[diouxX])?\$', _replace, tmpl)
|
||
|
||
|
||
def download_subs_as_srt(mpd_url, mpd_root, sub_tracks, sel_sub_langs, out_name, session):
|
||
"""Download subtitle tracks by fetching VTT segments, then convert to SRT.
|
||
|
||
Avoids ffmpeg's DASH demuxer (which mishandles presentationTimeOffset on
|
||
subtitle tracks). Instead, segment URLs are built from the MPD and each
|
||
segment is fetched via the authenticated session.
|
||
|
||
Plex naming convention:
|
||
{out_name}.{lang}.srt — regular subtitles
|
||
{out_name}.{lang}.forced.srt — forced subtitles (foreign dialogue only)
|
||
"""
|
||
if not sel_sub_langs:
|
||
return
|
||
|
||
# MPD base URL for resolving relative segment paths
|
||
mpd_no_qs = mpd_url.split('?')[0]
|
||
mpd_base = mpd_no_qs[:mpd_no_qs.rfind('/') + 1]
|
||
# Honour explicit <BaseURL> element if present
|
||
base_el = mpd_root.find(f'{MPD_NS}BaseURL')
|
||
if base_el is not None and base_el.text and base_el.text.startswith('http'):
|
||
mpd_base = base_el.text.rstrip('/') + '/'
|
||
|
||
# Collect subtitle AdaptationSet elements in MPD order
|
||
all_sub_as = [
|
||
el for el in mpd_root.iter(f'{MPD_NS}AdaptationSet')
|
||
if (el.get('contentType', '') == 'text' or
|
||
el.get('mimeType', '').startswith('text/'))
|
||
]
|
||
|
||
# Group selected tracks by language to determine forced/regular naming
|
||
lang_track_indices: dict = {} # lang → [sub_tracks index, ...]
|
||
for idx, (lang, _) in enumerate(sub_tracks):
|
||
if lang in sel_sub_langs:
|
||
lang_track_indices.setdefault(lang, []).append(idx)
|
||
|
||
print('Downloading subtitles...')
|
||
for lang, indices in lang_track_indices.items():
|
||
iso = _lang_tag_to_iso639_1(lang)
|
||
|
||
for pos, track_idx in enumerate(indices):
|
||
if track_idx >= len(all_sub_as):
|
||
continue
|
||
as_el = all_sub_as[track_idx]
|
||
|
||
if len(indices) > 1 and pos == 0:
|
||
label = f'{iso}.forced'
|
||
elif pos > 1:
|
||
label = f'{iso}.{pos}'
|
||
else:
|
||
label = iso
|
||
|
||
# SegmentTemplate may live on AdaptationSet or Representation
|
||
st = as_el.find(f'{MPD_NS}SegmentTemplate')
|
||
rep = next(as_el.iter(f'{MPD_NS}Representation'), None)
|
||
if st is None and rep is not None:
|
||
st = rep.find(f'{MPD_NS}SegmentTemplate')
|
||
if st is None:
|
||
print(f' Warning: no SegmentTemplate for {label}')
|
||
continue
|
||
|
||
media_tmpl = st.get('media', '')
|
||
start_num = int(st.get('startNumber', '1'))
|
||
rep_id = rep.get('id', '') if rep is not None else ''
|
||
|
||
# Build segment list from SegmentTimeline
|
||
timeline = st.find(f'{MPD_NS}SegmentTimeline')
|
||
segments = [] # list of (number, time)
|
||
if timeline is not None:
|
||
t = 0
|
||
num = start_num
|
||
for s_el in timeline:
|
||
t_attr = s_el.get('t')
|
||
if t_attr is not None:
|
||
t = int(t_attr)
|
||
d = int(s_el.get('d', 0))
|
||
r = int(s_el.get('r', 0))
|
||
for _ in range(r + 1):
|
||
segments.append((num, t))
|
||
t += d
|
||
num += 1
|
||
|
||
if not segments:
|
||
print(f' Warning: empty SegmentTimeline for {label}')
|
||
continue
|
||
|
||
# Download each VTT segment
|
||
vtt_parts = []
|
||
for seg_num, seg_t in segments:
|
||
url = _dash_segment_url(media_tmpl, rep_id, number=seg_num, time=seg_t)
|
||
if not url.startswith('http'):
|
||
url = mpd_base + url.lstrip('/')
|
||
try:
|
||
r = session.get(url, headers={'User-Agent': UA}, timeout=15)
|
||
if r.ok:
|
||
vtt_parts.append(r.text)
|
||
else:
|
||
print(f' Warning: segment {seg_num} HTTP {r.status_code}: {url}')
|
||
except Exception as exc:
|
||
print(f' Warning: segment {seg_num} fetch error: {exc}')
|
||
|
||
if len(vtt_parts) < len(segments):
|
||
print(f' Warning: [{label}] only {len(vtt_parts)}/{len(segments)} segments downloaded')
|
||
|
||
if not vtt_parts:
|
||
print(f' Warning: no segments downloaded for {label}')
|
||
continue
|
||
|
||
# Merge segments: keep WEBVTT header from first; append only cue
|
||
# blocks from subsequent segments (skip the WEBVTT/NOTE header lines).
|
||
merged_lines = vtt_parts[0].splitlines()
|
||
for part in vtt_parts[1:]:
|
||
lines = part.splitlines()
|
||
for i, line in enumerate(lines):
|
||
if '-->' in line:
|
||
# Back up one line to include optional cue identifier
|
||
start = (i - 1 if i > 0 and lines[i - 1].strip()
|
||
and '-->' not in lines[i - 1] else i)
|
||
merged_lines.append('')
|
||
merged_lines.extend(lines[start:])
|
||
break
|
||
|
||
srt_text = _vtt_to_srt('\n'.join(merged_lines))
|
||
cue_count = srt_text.count('\n\n') + (1 if srt_text.strip() else 0)
|
||
|
||
srt_path = SAVE_DIR / f'{out_name}.{label}.srt'
|
||
if srt_text.strip():
|
||
srt_path.write_text(srt_text, encoding='utf-8')
|
||
print(f' {srt_path.name} ({cue_count} cues)')
|
||
else:
|
||
print(f' Warning: no cues converted for {label}')
|
||
|
||
|
||
def download_and_decrypt(mpd_url, keys, out_name, video_args, audio_args):
|
||
key_args = []
|
||
for kid, key in keys.items():
|
||
key_args += ['--key', f'{kid}:{key}']
|
||
|
||
cmd = [
|
||
str(N_M3U8), mpd_url,
|
||
'--save-dir', str(SAVE_DIR),
|
||
'--save-name', out_name,
|
||
'--tmp-dir', TMP_DIR,
|
||
'--decryption-binary-path', str(MP4DEC),
|
||
'--mux-after-done', 'format=mkv',
|
||
'--header', f'User-Agent: {UA}',
|
||
'--drop-subtitle', 'all',
|
||
] + video_args + audio_args + key_args
|
||
|
||
print(f'N_m3u8DL-RE select: {" ".join(video_args + audio_args)}')
|
||
print('Downloading and decrypting...')
|
||
result = subprocess.run(cmd)
|
||
if result.returncode != 0:
|
||
raise RuntimeError(f'N_m3u8DL-RE failed (exit {result.returncode})')
|
||
|
||
|
||
def _slug_from_url(url):
|
||
"""Extract the ATOM catalog slug from a SkyShowtime watch URL.
|
||
|
||
https://www.skyshowtime.com/watch/asset/tv/name/uuid → /asset/tv/name/uuid
|
||
"""
|
||
from urllib.parse import urlparse
|
||
path = urlparse(url).path
|
||
if path.startswith('/watch'):
|
||
path = path[len('/watch'):]
|
||
return path
|
||
|
||
|
||
def _parse_episode_node(node):
|
||
"""Parse a single ASSET/EPISODE node into an episode dict, or None if unusable."""
|
||
att = node.get('attributes', {})
|
||
try:
|
||
season_num = int(att.get('seasonNumber', 1))
|
||
except (TypeError, ValueError):
|
||
season_num = 1
|
||
try:
|
||
ep_num = int(att.get('episodeNumber', att.get('number', 0)))
|
||
except (TypeError, ValueError):
|
||
ep_num = 0
|
||
ep_title = att.get('title', f'Episode {ep_num}')
|
||
|
||
content_id = None
|
||
for fmt in ('HD', 'SD'):
|
||
if fmt in att.get('formats', {}):
|
||
content_id = att['formats'][fmt].get('contentId')
|
||
if content_id:
|
||
break
|
||
provider_variant_id = att.get('providerVariantId')
|
||
if not content_id or not provider_variant_id:
|
||
return None
|
||
return {
|
||
'number': ep_num,
|
||
'title': ep_title,
|
||
'content_id': content_id,
|
||
'provider_variant_id': provider_variant_id,
|
||
'season_number': season_num,
|
||
}
|
||
|
||
|
||
def get_series_info(session, series_uuid):
|
||
"""Fetch series title and all available episodes from the ATOM catalog API.
|
||
|
||
Uses /node?slug=...&represent=(items(items)) which returns the full
|
||
series → season → episode tree in a single request.
|
||
|
||
Returns (series_title, episodes) sorted by (season_number, episode_number).
|
||
"""
|
||
# Step 1: resolve UUID → actual ATOM slug + title
|
||
uuid_url = (f'{ATOM}/adapter-calypso/v3/query/nodes/uuid/{series_uuid}'
|
||
f'?exclude=expired%2Cfuture%2Cshortform')
|
||
r = session.get(uuid_url, headers=SKY_HEADERS, timeout=15)
|
||
uuid_data = r.json()
|
||
if not uuid_data:
|
||
raise RuntimeError(f'No info returned for series UUID {series_uuid}')
|
||
series_node = uuid_data[0]
|
||
s_att = series_node.get('attributes', {})
|
||
series_title = s_att.get('title', series_uuid)
|
||
series_slug = s_att.get('slug', '')
|
||
if not series_slug:
|
||
raise RuntimeError(f'Series node has no slug (UUID {series_uuid})')
|
||
print(f'Series slug: {series_slug}')
|
||
|
||
# Step 2: fetch full series tree with nested season/episode data
|
||
url = (f'{ATOM}/adapter-calypso/v3/query/node'
|
||
f'?slug={series_slug}&represent=(items(items))')
|
||
r = session.get(url, headers=SKY_HEADERS, timeout=30)
|
||
data = r.json()
|
||
|
||
series_title = data.get('attributes', {}).get('title', series_title)
|
||
|
||
# Top-level items are seasons (CATALOGUE/SEASON) or episodes (ASSET/EPISODE)
|
||
top_items = (data.get('relationships', {})
|
||
.get('items', {})
|
||
.get('data', []))
|
||
|
||
if not top_items:
|
||
raise RuntimeError(f'No seasons/episodes found for series "{series_title}"')
|
||
|
||
all_episodes = []
|
||
|
||
for item in top_items:
|
||
node_type = item.get('type', '')
|
||
|
||
if node_type == 'ASSET/EPISODE':
|
||
ep = _parse_episode_node(item)
|
||
if ep:
|
||
all_episodes.append(ep)
|
||
|
||
elif node_type == 'CATALOGUE/SEASON':
|
||
s_att = item.get('attributes', {})
|
||
try:
|
||
season_num = int(s_att.get('seasonNumber', 1))
|
||
except (TypeError, ValueError):
|
||
season_num = 1
|
||
|
||
ep_nodes = (item.get('relationships', {})
|
||
.get('items', {})
|
||
.get('data', []))
|
||
|
||
if ep_nodes:
|
||
for ep_node in ep_nodes:
|
||
ep = _parse_episode_node(ep_node)
|
||
if ep:
|
||
ep['season_number'] = season_num
|
||
all_episodes.append(ep)
|
||
else:
|
||
# represent=(items(items)) didn't populate nested episodes;
|
||
# fall back to a separate request using the season slug.
|
||
season_slug = s_att.get('slug', '')
|
||
if season_slug:
|
||
ep_r = session.get(
|
||
f'{ATOM}/adapter-calypso/v3/query/node'
|
||
f'?slug={season_slug}&represent=(items)',
|
||
headers=SKY_HEADERS, timeout=15)
|
||
ep_data = ep_r.json()
|
||
ep_nodes2 = (ep_data.get('relationships', {})
|
||
.get('items', {})
|
||
.get('data', []))
|
||
for ep_node in ep_nodes2:
|
||
ep = _parse_episode_node(ep_node)
|
||
if ep:
|
||
ep['season_number'] = season_num
|
||
all_episodes.append(ep)
|
||
|
||
all_episodes.sort(key=lambda e: (e['season_number'], e['number']))
|
||
return series_title, all_episodes
|
||
|
||
|
||
def select_episode_interactive(series_title, episodes):
|
||
"""Single-select episode picker with an [All episodes] option at the top.
|
||
|
||
Returns the list of episode dicts to download (all, or a single one).
|
||
"""
|
||
items = ['[All episodes]'] + [
|
||
f'S{e["season_number"]:02d}E{e["number"]:02d} {e["title"]}'
|
||
for e in episodes
|
||
]
|
||
print(f'\nSeries: {series_title} ({len(episodes)} episodes)')
|
||
sel = checkbox_picker('Select episode to download', items,
|
||
single_select=True, default_idx=0)
|
||
idx = sel[0] if sel else 0
|
||
if idx == 0:
|
||
print('Downloading all episodes.')
|
||
return list(episodes)
|
||
chosen = episodes[idx - 1]
|
||
print(f'Downloading S{chosen["season_number"]:02d}E{chosen["number"]:02d}: {chosen["title"]}')
|
||
return [chosen]
|
||
|
||
|
||
def episode_out_name(ep):
|
||
"""Plex-compatible episode filename: s01e03_Title_Words"""
|
||
return f's{ep["season_number"]:02d}e{ep["number"]:02d}_{safe_filename(ep["title"])}'
|
||
|
||
|
||
def download_episode(session, user_token, ep, video_args, audio_args, sel_sub_langs):
|
||
"""Fetch playback info, decrypt, and save one episode with its subtitles."""
|
||
s, e = ep['season_number'], ep['number']
|
||
print(f'\n── S{s:02d}E{e:02d}: {ep["title"]} ──')
|
||
|
||
manifest_url, license_url = get_playback_info(
|
||
session, user_token, ep['content_id'], ep['provider_variant_id']
|
||
)
|
||
|
||
_, mpd_root = fetch_mpd(manifest_url, session)
|
||
pssh_b64 = get_pssh_from_root(mpd_root)
|
||
_, _, _, sub_tracks = parse_mpd_tracks(mpd_root)
|
||
|
||
print('Fetching content key...')
|
||
keys = fetch_content_key(pssh_b64, license_url)
|
||
for kid, key in keys.items():
|
||
print(f' {kid}:{key}')
|
||
|
||
out_name = episode_out_name(ep)
|
||
download_and_decrypt(manifest_url, keys, out_name, video_args, audio_args)
|
||
if sel_sub_langs:
|
||
download_subs_as_srt(manifest_url, mpd_root, sub_tracks,
|
||
sel_sub_langs, out_name, session)
|
||
print(f'Saved: {SAVE_DIR / out_name}.mkv')
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(
|
||
description='Download a DRM-protected video from SkyShowtime',
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""
|
||
Examples:
|
||
# Interactive video/audio/subtitle selection (default when running in terminal)
|
||
python3 sky_download.py <url>
|
||
|
||
# All audio + all subtitles, best video (non-interactive)
|
||
python3 sky_download.py -a all -s all <url>
|
||
|
||
# Polish + English audio, Polish subtitles, specific resolution
|
||
python3 sky_download.py -a pl-PL,en-US -s pl-PL -v 1920x1080 <url>
|
||
|
||
# No subtitles
|
||
python3 sky_download.py -s none <url>
|
||
|
||
# Credentials via env vars (avoids password in shell history)
|
||
export SKY_USER=me@email.com
|
||
read -rs SKY_PASS && export SKY_PASS
|
||
python3 sky_download.py <url>
|
||
""",
|
||
)
|
||
parser.add_argument('url', help='https://www.skyshowtime.com/watch/... URL')
|
||
parser.add_argument('-u', '--username', default=os.environ.get('SKY_USER'),
|
||
help='SkyShowtime email (or set SKY_USER env var)')
|
||
parser.add_argument('-p', '--password', default=os.environ.get('SKY_PASS'),
|
||
help='SkyShowtime password (or set SKY_PASS env var)')
|
||
parser.add_argument('-o', '--output',
|
||
help='Output filename without extension. Default: video title.')
|
||
parser.add_argument('-v', '--video', default=None, metavar='RES',
|
||
help=('Video quality: resolution like "1920x1080", "best", or "worst". '
|
||
'Omit for interactive selection.'))
|
||
parser.add_argument('-a', '--audio', default=None, metavar='LANGS',
|
||
help=('Audio tracks: comma-separated lang codes (e.g. pl-PL,en-US), '
|
||
'"all", or "none". Omit for interactive selection.'))
|
||
parser.add_argument('-s', '--subs', default=None, metavar='LANGS',
|
||
help=('Subtitle tracks: comma-separated lang codes, "all", or "none". '
|
||
'Omit for interactive selection.'))
|
||
args = parser.parse_args()
|
||
|
||
if not WVD:
|
||
sys.exit('ERROR: No valid .wvd file found in L3/.')
|
||
|
||
session = requests.Session()
|
||
|
||
# ── Session restore ───────────────────────────────────────────────────────
|
||
# Persist cookie, device_id, profile_id AND user_token across runs.
|
||
# The /auth/throttled/tokens call is what registers a new device in the
|
||
# SkyShowtime account — so we cache user_token and only refresh it when
|
||
# a playback request tells us it has expired.
|
||
saved_session: dict = {}
|
||
if SESSION_FILE.exists():
|
||
try:
|
||
saved_session = json.loads(SESSION_FILE.read_text())
|
||
except Exception:
|
||
pass
|
||
|
||
# Always reuse the same device_id so no "new device" notification is triggered.
|
||
device_id = (saved_session.get('device_id') or
|
||
''.join(random.choices(string.ascii_letters + string.digits, k=20)))
|
||
|
||
cookie_str = saved_session.get('cookie_str', '')
|
||
profile_id = saved_session.get('profile_id', '')
|
||
user_token = saved_session.get('user_token', '')
|
||
|
||
def _refresh_token():
|
||
"""Fetch a new user_token (registers device) and persist the session."""
|
||
nonlocal user_token
|
||
print('Refreshing user token...')
|
||
user_token = get_user_token(session, cookie_str, profile_id, device_id)
|
||
_save_session()
|
||
print(f'User token: {user_token[:30]}...')
|
||
|
||
def _save_session():
|
||
SESSION_FILE.write_text(json.dumps({
|
||
'cookie_str': cookie_str,
|
||
'device_id': device_id,
|
||
'profile_id': profile_id,
|
||
'user_token': user_token,
|
||
}))
|
||
SESSION_FILE.chmod(0o600)
|
||
|
||
if user_token and cookie_str and profile_id:
|
||
print('Using cached session.')
|
||
else:
|
||
# No usable session — need a full login.
|
||
if not args.username or not args.password:
|
||
if CREDS_FILE.exists():
|
||
saved_creds = json.loads(CREDS_FILE.read_text())
|
||
args.username = args.username or saved_creds.get('username', '')
|
||
args.password = args.password or saved_creds.get('password', '')
|
||
|
||
prompted = False
|
||
if not args.username:
|
||
args.username = input('SkyShowtime username (email): ').strip()
|
||
prompted = True
|
||
if not args.password:
|
||
args.password = getpass.getpass('SkyShowtime password: ')
|
||
prompted = True
|
||
|
||
print(f'Logging in as {args.username}...')
|
||
try:
|
||
cookie_str = login(session, args.username, args.password)
|
||
except RuntimeError as e:
|
||
if CREDS_FILE.exists() and not prompted:
|
||
print(f'ERROR: {e}')
|
||
print(f'Hint: saved credentials may be wrong. Delete {CREDS_FILE} to re-enter them.')
|
||
sys.exit(1)
|
||
raise
|
||
print('Logged in.')
|
||
|
||
if prompted:
|
||
CREDS_FILE.write_text(json.dumps({'username': args.username,
|
||
'password': args.password}))
|
||
CREDS_FILE.chmod(0o600)
|
||
print(f'Credentials saved to {CREDS_FILE}')
|
||
|
||
print('Getting profile...')
|
||
profile_id = get_profile_id(session, cookie_str)
|
||
print(f'Profile ID: {profile_id[:20]}...')
|
||
|
||
_refresh_token() # first-time token fetch; saves session
|
||
|
||
_save_session()
|
||
# ── End session restore ───────────────────────────────────────────────────
|
||
|
||
territory = detect_territory(cookie_str)
|
||
if territory:
|
||
SKY_HEADERS['x-skyott-territory'] = territory
|
||
SKY_HEADERS['x-skyott-activeterritory'] = territory
|
||
print(f'Territory: {territory}')
|
||
|
||
uuid = extract_uuid(args.url)
|
||
print(f'UUID: {uuid}')
|
||
|
||
is_series = '/watch/asset/tv/' in args.url
|
||
|
||
use_interactive = (args.video is None and args.audio is None
|
||
and args.subs is None and sys.stdin.isatty())
|
||
|
||
if is_series:
|
||
# ── Series flow ──────────────────────────────────────────────────────
|
||
print('Getting series info...')
|
||
series_title, all_episodes = get_series_info(session, uuid)
|
||
if not all_episodes:
|
||
sys.exit('ERROR: No downloadable episodes found for this series.')
|
||
|
||
selected_eps = select_episode_interactive(series_title, all_episodes)
|
||
|
||
# Fetch the first episode's MPD to offer track selection that applies
|
||
# to every episode in the batch.
|
||
first_ep = selected_eps[0]
|
||
print(f'\nFetching track info from '
|
||
f'S{first_ep["season_number"]:02d}E{first_ep["number"]:02d}: '
|
||
f'{first_ep["title"]}...')
|
||
try:
|
||
manifest_url_first, _ = get_playback_info(
|
||
session, user_token, first_ep['content_id'], first_ep['provider_variant_id'])
|
||
except TokenExpiredError:
|
||
_refresh_token()
|
||
manifest_url_first, _ = get_playback_info(
|
||
session, user_token, first_ep['content_id'], first_ep['provider_variant_id'])
|
||
_, mpd_root_first = fetch_mpd(manifest_url_first, session)
|
||
video_streams, audio_tracks, sub_langs, _ = parse_mpd_tracks(mpd_root_first)
|
||
|
||
if use_interactive:
|
||
video_args = select_video_interactive(video_streams)
|
||
audio_args, sel_sub_langs = select_tracks_interactive(audio_tracks, sub_langs)
|
||
else:
|
||
v = args.video
|
||
if v and v not in ('best', 'worst'):
|
||
video_args = ['--select-video', f'res={v}:for=best']
|
||
else:
|
||
video_args = ['--select-video', v or 'best']
|
||
audio_spec = args.audio or 'all'
|
||
audio_args = cli_spec_to_select_args('--select-audio', '--drop-audio', audio_spec)
|
||
subs_spec = args.subs or 'all'
|
||
if subs_spec.lower() == 'none':
|
||
sel_sub_langs = set()
|
||
elif subs_spec.lower() == 'all':
|
||
sel_sub_langs = set(sub_langs.keys())
|
||
else:
|
||
sel_sub_langs = set(subs_spec.split(','))
|
||
|
||
SAVE_DIR.mkdir(exist_ok=True)
|
||
print(f'\nDownloading {len(selected_eps)} episode(s)...')
|
||
for ep in selected_eps:
|
||
try:
|
||
download_episode(session, user_token, ep, video_args, audio_args, sel_sub_langs)
|
||
except TokenExpiredError:
|
||
_refresh_token()
|
||
download_episode(session, user_token, ep, video_args, audio_args, sel_sub_langs)
|
||
print(f'\nAll done. Files saved to {SAVE_DIR}')
|
||
|
||
else:
|
||
# ── Movie / single-item flow ──────────────────────────────────────────
|
||
print('Getting video info...')
|
||
title, content_id, provider_variant_id = get_video_info(session, uuid)
|
||
print(f'Title: {title}')
|
||
print(f'Content ID: {content_id}')
|
||
print(f'Provider Variant ID: {provider_variant_id}')
|
||
|
||
print('Getting playback info...')
|
||
try:
|
||
manifest_url, license_url = get_playback_info(
|
||
session, user_token, content_id, provider_variant_id)
|
||
except TokenExpiredError:
|
||
_refresh_token()
|
||
manifest_url, license_url = get_playback_info(
|
||
session, user_token, content_id, provider_variant_id)
|
||
print(f'Manifest: {manifest_url[:80]}...')
|
||
print(f'License: {license_url[:80]}...')
|
||
|
||
print('Fetching MPD...')
|
||
_, mpd_root = fetch_mpd(manifest_url, session)
|
||
|
||
print('Extracting PSSH...')
|
||
pssh_b64 = get_pssh_from_root(mpd_root)
|
||
print(f'PSSH: {pssh_b64[:48]}...')
|
||
|
||
video_streams, audio_tracks, sub_langs, sub_tracks = parse_mpd_tracks(mpd_root)
|
||
|
||
if use_interactive:
|
||
video_args = select_video_interactive(video_streams)
|
||
audio_args, sel_sub_langs = select_tracks_interactive(audio_tracks, sub_langs)
|
||
else:
|
||
v = args.video
|
||
if v and v not in ('best', 'worst'):
|
||
video_args = ['--select-video', f'res={v}:for=best']
|
||
else:
|
||
video_args = ['--select-video', v or 'best']
|
||
audio_spec = args.audio or 'all'
|
||
audio_args = cli_spec_to_select_args('--select-audio', '--drop-audio', audio_spec)
|
||
subs_spec = args.subs or 'all'
|
||
if subs_spec.lower() == 'none':
|
||
sel_sub_langs = set()
|
||
elif subs_spec.lower() == 'all':
|
||
sel_sub_langs = set(sub_langs.keys())
|
||
else:
|
||
sel_sub_langs = set(subs_spec.split(','))
|
||
|
||
print('Fetching content key...')
|
||
keys = fetch_content_key(pssh_b64, license_url)
|
||
if not keys:
|
||
raise RuntimeError('License server returned no content keys.')
|
||
for kid, key in keys.items():
|
||
print(f' {kid}:{key}')
|
||
|
||
SAVE_DIR.mkdir(exist_ok=True)
|
||
out_name = args.output if args.output else safe_filename(title)
|
||
|
||
download_and_decrypt(manifest_url, keys, out_name, video_args, audio_args)
|
||
if sel_sub_langs:
|
||
download_subs_as_srt(manifest_url, mpd_root, sub_tracks,
|
||
sel_sub_langs, out_name, session)
|
||
print(f'Done: {SAVE_DIR / out_name}.mkv')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|