mirror of
https://github.com/n0stal6ic/unshackle-services.git
synced 2026-07-15 17:40:02 +02:00
2257 lines
100 KiB
Python
2257 lines
100 KiB
Python
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import random
|
|
import re
|
|
import secrets
|
|
import string
|
|
import time
|
|
import uuid
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
from http.cookiejar import CookieJar
|
|
from typing import Any, Optional
|
|
from urllib.parse import quote
|
|
import click
|
|
import requests
|
|
from click.core import ParameterSource
|
|
from langcodes import Language
|
|
from tldextract import tldextract
|
|
from unshackle.core.cacher import Cacher
|
|
from unshackle.core.credential import Credential
|
|
from unshackle.core.manifests import DASH, ISM
|
|
from unshackle.core.service import Service
|
|
from unshackle.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
|
|
from unshackle.core.tracks import Attachment, Chapter, Chapters, Subtitle, Tracks, Video
|
|
from unshackle.core.utils.collections import as_list
|
|
|
|
|
|
def _build_ordered_lang_map_from_mpd(mpd_text: str) -> dict:
|
|
import xml.etree.ElementTree as ET
|
|
import re as _re
|
|
ns_strip = _re.compile(r"\{[^}]*\}")
|
|
rid_lang_re = _re.compile(r"^audio_([a-zA-Z]{2,3}-[a-zA-Z0-9]{2,5})")
|
|
result: dict = {}
|
|
try:
|
|
root = ET.fromstring(mpd_text)
|
|
for elem in root.iter():
|
|
if ns_strip.sub("", elem.tag) != "AdaptationSet":
|
|
continue
|
|
content_type = elem.get("contentType", "") or elem.get("mimeType", "")
|
|
if "audio" not in content_type.lower():
|
|
if not any("audio" in (r.get("mimeType", "")).lower() for r in elem):
|
|
continue
|
|
base_lang = elem.get("lang") or elem.get("language") or ""
|
|
if not base_lang:
|
|
continue
|
|
for r in elem:
|
|
if ns_strip.sub("", r.tag) != "Representation":
|
|
continue
|
|
rid = r.get("id") or ""
|
|
m = rid_lang_re.match(rid)
|
|
precise = m.group(1) if m else base_lang
|
|
result.setdefault(base_lang, []).append(precise)
|
|
except Exception:
|
|
pass
|
|
return result
|
|
|
|
|
|
def _apply_ordered_lang_map(audio_tracks, lang_map: dict) -> None:
|
|
from langcodes import Language as _Lang
|
|
counters: dict = {}
|
|
for track in audio_tracks:
|
|
base = str(track.language)
|
|
if base not in lang_map:
|
|
continue
|
|
ordered = lang_map[base]
|
|
if not any("-" in p for p in ordered):
|
|
continue
|
|
idx = counters.get(base, 0)
|
|
if idx < len(ordered):
|
|
track.language = _Lang.get(ordered[idx])
|
|
counters[base] = idx + 1
|
|
|
|
|
|
def _resolve_subtitle_language(language_code: str, url: str) -> str:
|
|
if "-" in language_code:
|
|
return language_code
|
|
import re as _re
|
|
m = _re.search(
|
|
r"(?:^|[/_-])(" + _re.escape(language_code) + r"[-_][A-Za-z0-9]{2,5})(?:[/_.-]|$)",
|
|
url, _re.IGNORECASE
|
|
)
|
|
if m:
|
|
return m.group(1).replace("_", "-")
|
|
return language_code
|
|
|
|
|
|
class AMZN(Service):
|
|
"""
|
|
Service code for Amazon VOD (https://amazon.com) & Amazon Prime Video (https://primevideo.com).
|
|
|
|
www.nostalgic.cc
|
|
Authorization: Credentials, Cookies
|
|
Security: UHD@L1 FHD@Chrome SD@L3
|
|
"""
|
|
|
|
ALIASES = ["AMZN", "amazon", "prime"]
|
|
TITLE_RE = [
|
|
r"^(?:https?://(?:www\.)?(?P<domain>amazon\.(?P<region>com|co\.uk|de|co\.jp)|primevideo\.com)(?:/.+)?/)?(?P<id>[A-Z0-9]{10,}|amzn1\.dv\.gti\.[a-f0-9-]+)",
|
|
r"^(?:https?://(?:www\.)?(?P<domain>amazon\.(?P<region>com|co\.uk|de|co\.jp)|primevideo\.com)(?:/[^?]*)?(?:\?gti=)?)(?P<id>[A-Z0-9]{10,}|amzn1\.dv\.gti\.[a-f0-9-]+)"
|
|
]
|
|
|
|
REGION_TLD_MAP = {
|
|
"au": "com.au",
|
|
"br": "com.br",
|
|
"jp": "co.jp",
|
|
"mx": "com.mx",
|
|
"tr": "com.tr",
|
|
"gb": "co.uk",
|
|
"us": "com",
|
|
}
|
|
VIDEO_RANGE_MAP = {
|
|
"SDR": "None",
|
|
"HDR10": "Hdr10",
|
|
"DV": "DolbyVision",
|
|
}
|
|
|
|
@staticmethod
|
|
@click.command(name="AMZN", short_help="https://amazon.com, https://primevideo.com", help=__doc__)
|
|
@click.argument("title", type=str, required=False)
|
|
@click.option("-b", "--bitrate", default="CBR",
|
|
type=click.Choice(["CVBR", "CBR", "CVBR+CBR"], case_sensitive=False),
|
|
help="Video Bitrate Mode to download in. CVBR=Constrained Variable Bitrate, CBR=Constant Bitrate.")
|
|
@click.option("-p", "--player", default="html5",
|
|
type=click.Choice(["html5", "xp"], case_sensitive=False),
|
|
help="Video playerType to download in. html5, xp.")
|
|
@click.option("-c", "--cdn", default=None, type=str,
|
|
help="CDN to download from, defaults to a random CDN from the available set.")
|
|
@click.option("-vq", "--vquality", default="HD",
|
|
type=click.Choice(["SD", "HD", "UHD"], case_sensitive=False),
|
|
help="Manifest quality to request.")
|
|
@click.option("-s", "--single", is_flag=True, default=False,
|
|
help="Force single episode/season instead of getting series ASIN.")
|
|
@click.option("-am", "--amanifest", default="CVBR",
|
|
type=click.Choice(["CVBR", "CBR", "H265"], case_sensitive=False),
|
|
help="Manifest to use for audio. Defaults to H265 if the video manifest is missing 640k audio.")
|
|
@click.option("-aq", "--aquality", default="SD",
|
|
type=click.Choice(["SD", "HD", "UHD"], case_sensitive=False),
|
|
help="Manifest quality to request for audio. Defaults to the same as --quality.")
|
|
@click.option("-mt", "--manifest", "manifest_type", default="DASH",
|
|
type=click.Choice(["DASH", "ISM"], case_sensitive=False),
|
|
help="Manifest protocol for the video stream. ISM (SmoothStreaming) can carry a "
|
|
"higher-bitrate HDR10 video track than DASH, but requires a registered device.")
|
|
@click.option("-drm", "--drm-system", type=click.Choice(["widevine", "playready"], case_sensitive=False),
|
|
default="playready",
|
|
help="which drm system to use")
|
|
@click.option("-nr", "--no-true-region", "no_true_region", is_flag=True, default=False,
|
|
help="Skip the 'true region' playback session (keep-alive) and don't adopt the IP-geo "
|
|
"marketplace from the configuration endpoint. Session keep-alive is ON by default.")
|
|
@click.option("-pl", "--playlisted", is_flag=True, default=False,
|
|
help="Use Amazon's newer LivingRoomPlayer 'vodPlaylistedPlaybackUrls' request path "
|
|
"(requires a registered device). Off by default; the classic path is used first and "
|
|
"this is also tried automatically as a fallback when a device is registered.")
|
|
@click.pass_context
|
|
def cli(ctx, **kwargs):
|
|
return AMZN(ctx, **kwargs)
|
|
|
|
def __init__(self, ctx, title, bitrate: str, player: str, cdn: str, vquality: str, single: bool,
|
|
amanifest: str, aquality: str, manifest_type: str, drm_system: str,
|
|
no_true_region: bool, playlisted: bool):
|
|
super().__init__(ctx)
|
|
self.parse_title(ctx, title)
|
|
self.bitrate = bitrate
|
|
self.player = player
|
|
self.bitrate_source = ctx.get_parameter_source("bitrate")
|
|
self.cdn = cdn
|
|
self.vquality = vquality
|
|
self.vquality_source = ctx.get_parameter_source("vquality")
|
|
self.single = single
|
|
self.amanifest = amanifest
|
|
self.aquality = aquality
|
|
self.manifest_type = (manifest_type or "DASH").upper()
|
|
self.manifest_type_source = ctx.get_parameter_source("manifest_type")
|
|
self.drm_system = drm_system
|
|
self.no_true_region = no_true_region
|
|
self.playlisted = playlisted
|
|
|
|
assert ctx.parent is not None
|
|
self.ctx = ctx
|
|
|
|
self.chapters_only = ctx.parent.params.get("chapters_only")
|
|
self.quality = ctx.parent.params.get("quality") or 1080
|
|
|
|
vcodec = ctx.parent.params.get("vcodec")
|
|
range_ = ctx.parent.params.get("range_")
|
|
|
|
self.range = range_[0].name if range_ else "SDR"
|
|
vstr = str(vcodec).upper() if vcodec else ""
|
|
if "AV1" in vstr:
|
|
self.vcodec = "AV1"
|
|
elif "HEVC" in vstr:
|
|
self.vcodec = "H265"
|
|
elif "AVC" in vstr:
|
|
self.vcodec = "H264"
|
|
else:
|
|
self.vcodec = "AV1"
|
|
|
|
self.cdm = ctx.obj.cdm
|
|
self.profile = ctx.obj.profile
|
|
self.playready = self.drm_system == "playready"
|
|
if ctx.get_parameter_source("drm_system") != ParameterSource.COMMANDLINE:
|
|
try:
|
|
from unshackle.core.cdm.detect import is_playready_cdm, is_widevine_cdm
|
|
if is_widevine_cdm(self.cdm):
|
|
self.playready = False
|
|
elif is_playready_cdm(self.cdm):
|
|
self.playready = True
|
|
except Exception:
|
|
pass
|
|
self.log.info(f" + DRM system: {'PlayReady' if self.playready else 'Widevine'}")
|
|
|
|
self.region: dict[str, str] = {}
|
|
self.endpoints: dict[str, str] = {}
|
|
self.device: dict[str, str] = {}
|
|
|
|
self.pv = False
|
|
self.event = False
|
|
self.device_token = None
|
|
self.device_refresh_token = None
|
|
self.device_id = None
|
|
self.customer_id = None
|
|
self.client_id = "f22dbddb-ef2c-48c5-8876-bed0d47594fd"
|
|
self.playbackEnvelope = None
|
|
self.playbackInfo = None
|
|
self.session_handoff_token = None
|
|
self.actor_token = None
|
|
self.profile_id = None
|
|
self.living_room = False
|
|
|
|
if self.vquality_source != ParameterSource.COMMANDLINE:
|
|
q_check = self.quality[0] if isinstance(self.quality, list) else self.quality
|
|
|
|
if 0 < q_check <= 576 and self.range == "SDR":
|
|
self.log.info(" + Setting manifest quality to SD")
|
|
self.vquality = "SD"
|
|
|
|
if q_check > 1080:
|
|
self.log.info(" + Setting manifest quality to UHD to be able to get 2160p video track")
|
|
self.vquality = "UHD"
|
|
|
|
self.vquality = self.vquality or "HD"
|
|
|
|
if self.vquality == "UHD":
|
|
self.vcodec = "H265"
|
|
|
|
if self.bitrate_source != ParameterSource.COMMANDLINE:
|
|
if self.manifest_type == "ISM":
|
|
self.bitrate = "CBR"
|
|
self.log.info(" + Forcing bitrate mode to CBR for ISM (SmoothStreaming)")
|
|
elif self.vcodec == "AV1":
|
|
self.bitrate = "CVBR"
|
|
self.log.info(" + Forcing bitrate mode to CVBR for AV1 codec")
|
|
elif self.vcodec == "H265" and self.range == "SDR" and self.bitrate != "CVBR+CBR":
|
|
self.bitrate = "CVBR+CBR"
|
|
self.log.info(" + Changed bitrate mode to CVBR+CBR to be able to get H.265 SDR video track")
|
|
|
|
if self.manifest_type != "ISM" and self.vquality == "UHD" and self.range != "SDR" and self.bitrate != "CBR":
|
|
self.bitrate = "CBR"
|
|
self.log.info(f" + Changed bitrate mode to CBR to be able to get highest quality UHD {self.range} video track")
|
|
|
|
self.orig_bitrate = self.bitrate
|
|
self.requested_vcodec = self.vcodec
|
|
|
|
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
|
|
super().authenticate(cookies, credential)
|
|
if not cookies:
|
|
raise EnvironmentError("Service requires Cookies for Authentication.")
|
|
|
|
self.session.cookies.update(cookies)
|
|
self.configure()
|
|
|
|
def configure(self) -> None:
|
|
if len(self.title) > 10:
|
|
self.pv = True
|
|
self.pv = True
|
|
|
|
self.log.info("Getting Account Region")
|
|
self.region = self.get_region()
|
|
if not self.region:
|
|
self.log.error(" - Failed to get Amazon Account region"); raise SystemExit(1)
|
|
|
|
self.log.info(f" + Region: {self.region['code']}")
|
|
|
|
self.endpoints = self.prepare_endpoints(self.config["endpoints"], self.region)
|
|
|
|
self.session.headers.update({
|
|
"Origin": f"https://{self.region['base']}"
|
|
})
|
|
|
|
_profile = self.profile or "default"
|
|
self.device = (self.config.get("device") or {}).get(_profile, {})
|
|
|
|
dtid_dict = self.config.get("dtid_dict", [])
|
|
if self.device and dtid_dict:
|
|
if self.device.get("device_type") not in set(dtid_dict):
|
|
self.log.error(
|
|
f" - Device type '{self.device.get('device_type')}' is NOT in the approved "
|
|
"dtid_dict. Using it could result in an Amazon account ban. Update your config."
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
need_device = False
|
|
if (isinstance(self.quality, list) and self.quality[0] > 1080) or self.vquality == "UHD" or self.range != "SDR":
|
|
need_device = True
|
|
if self.manifest_type == "ISM":
|
|
need_device = True
|
|
if self.playlisted:
|
|
need_device = True
|
|
|
|
if self.manifest_type == "ISM" and not self.device:
|
|
self.log.error(
|
|
" - ISM/SmoothStreaming requires a configured device in config.yaml (device: profile). "
|
|
"Add one or use --manifest DASH."
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
if self.playlisted and not self.device:
|
|
self.log.warning(
|
|
" - --playlisted (LivingRoomPlayer) needs a configured device in config.yaml; "
|
|
"falling back to the classic browser path."
|
|
)
|
|
self.playlisted = False
|
|
|
|
if self.device:
|
|
if need_device and self.vcodec == "H265":
|
|
self.log.info("Using device to get UHD manifests")
|
|
else:
|
|
self.log.info(f"Using configured device for profile: {_profile}")
|
|
res_cfg = self.session.get(
|
|
url=self.endpoints["configuration"],
|
|
params={"deviceTypeID": self.device["device_type"], "deviceID": "Tv"}
|
|
)
|
|
self._apply_verified_territory(res_cfg)
|
|
self.register_device()
|
|
else:
|
|
self.log.warning(
|
|
"No Device information was provided for %s, using browser device...",
|
|
self.profile
|
|
)
|
|
self.device_id = hashlib.sha224(
|
|
("CustomerID" + self.session.headers["User-Agent"]).encode("utf-8")
|
|
).hexdigest()
|
|
self.device = {"device_type": self.config["device_types"]["browser"]}
|
|
res_cfg = self.session.get(
|
|
url=self.endpoints["configuration"],
|
|
params={"deviceTypeID": self.device["device_type"], "deviceID": "Web"}
|
|
)
|
|
self._apply_verified_territory(res_cfg)
|
|
if self.playlisted and self.device_token:
|
|
if self._ensure_actor_token():
|
|
self.living_room = True
|
|
self.log.info(" + LivingRoomPlayer mode active (actor-token envelope + licensing)")
|
|
else:
|
|
self.log.warning(
|
|
" - Could not acquire an actor token; --playlisted will use the classic "
|
|
"envelope/licensing context"
|
|
)
|
|
|
|
def _apply_verified_territory(self, res_cfg) -> None:
|
|
if res_cfg.status_code != 200:
|
|
self.log.warning(f" - Configuration endpoint returned {res_cfg.status_code}, using config values")
|
|
return
|
|
try:
|
|
ctx = res_cfg.json().get("requestContext", {})
|
|
except ValueError:
|
|
return
|
|
territory = ctx.get("currentTerritory")
|
|
marketplace = ctx.get("marketplaceID")
|
|
account_region = self.region.get("code")
|
|
if territory and account_region and territory.lower() != account_region.lower():
|
|
self.log.warning(
|
|
f" - IP region '{territory}' does not match account region '{account_region}'. "
|
|
f"Amazon geo-locates by IP — licensing (and sometimes title lookup) may be blocked. "
|
|
f"If the license step fails, use a {account_region} IP (e.g. --proxy {account_region} or a matching VPN)."
|
|
)
|
|
return
|
|
if territory:
|
|
self.log.info(f" + Region (verified): {territory}")
|
|
if marketplace and not self.no_true_region:
|
|
self.region["marketplace_id"] = marketplace
|
|
elif marketplace and self.no_true_region:
|
|
self.log.debug(" + --no-true-region set; keeping marketplace from config, ignoring IP-geo value")
|
|
|
|
@staticmethod
|
|
def _clean_show_name(*candidates: Optional[str]) -> str:
|
|
season_suffix = re.compile(
|
|
r"[\s:\-–—]+(?:Season|Series|Staffel|Saison|Temporada|Stagione|Seizoen)\s+\d+\s*$",
|
|
re.IGNORECASE,
|
|
)
|
|
cleaned_fallback = ""
|
|
for candidate in candidates:
|
|
if not candidate:
|
|
continue
|
|
cleaned = season_suffix.sub("", candidate).strip()
|
|
if cleaned and cleaned == candidate.strip():
|
|
return cleaned
|
|
if cleaned and not cleaned_fallback:
|
|
cleaned_fallback = cleaned
|
|
if cleaned_fallback:
|
|
return cleaned_fallback
|
|
return next((c.strip() for c in candidates if c), "")
|
|
|
|
def get_titles(self) -> Titles_T:
|
|
res = self.session.get(
|
|
url=self.endpoints["details"],
|
|
params={
|
|
"titleID": self.title,
|
|
"isElcano": "1",
|
|
"sections": ["Atf", "Btf"]
|
|
},
|
|
headers={"Accept": "application/json"}
|
|
)
|
|
|
|
if not res.ok:
|
|
self.log.error(f"Unable to get title: {res.text} [{res.status_code}]"); raise SystemExit(1)
|
|
|
|
data = res.json()["widgets"]
|
|
product_details = data.get("productDetails", {}).get("detail")
|
|
|
|
if not product_details:
|
|
error = res.json()["degradations"][0]
|
|
self.log.error(f"Unable to get title: {error['message']} [{error['code']}]"); raise SystemExit(1)
|
|
|
|
if data["pageContext"]["subPageType"] == "Event":
|
|
self.event = True
|
|
|
|
if data["pageContext"]["subPageType"] in ["Movie", "Event"]:
|
|
card = data["productDetails"]["detail"]
|
|
return Movies([Movie(
|
|
id_=card["catalogId"],
|
|
name=product_details["title"],
|
|
year=card.get("releaseYear", ""),
|
|
service=self.__class__,
|
|
data=card
|
|
)])
|
|
|
|
episodes_list = []
|
|
seasons = []
|
|
for season in (data.get("seasonSelector") or []):
|
|
sid = season.get("titleID")
|
|
if sid and sid not in seasons:
|
|
seasons.append(sid)
|
|
if self.single:
|
|
break
|
|
if not seasons and not self.single:
|
|
seasons = self._discover_seasons_via_spa()
|
|
if not seasons:
|
|
seasons = [self.title]
|
|
|
|
for season in seasons:
|
|
res = self.session.get(
|
|
url=self.endpoints["details"],
|
|
params={"titleID": season, "isElcano": "1", "sections": "Btf"},
|
|
headers={"Accept": "application/json"},
|
|
).json()["widgets"]
|
|
|
|
try:
|
|
episode_data_list = res["episodeList"]["episodes"]
|
|
except KeyError:
|
|
continue
|
|
|
|
product_details_season = res["productDetails"]["detail"]
|
|
show_name = self._clean_show_name(
|
|
product_details_season.get("parentTitle"),
|
|
product_details.get("parentTitle"),
|
|
product_details_season.get("title"),
|
|
product_details.get("title"),
|
|
)
|
|
|
|
for episode in episode_data_list:
|
|
details = episode["detail"]
|
|
episodes_list.append(Episode(
|
|
id_=details["catalogId"],
|
|
title=show_name,
|
|
name=details["title"],
|
|
season=product_details_season["seasonNumber"],
|
|
number=episode["self"]["sequenceNumber"],
|
|
service=self.__class__,
|
|
data=episode
|
|
))
|
|
|
|
pagination_data = res.get("episodeList", {}).get("actions", {}).get("pagination", [])
|
|
token = next((quote(item.get("token")) for item in pagination_data if item.get("tokenType") == "NextPage"), None)
|
|
|
|
while token:
|
|
res_page = self.session.get(
|
|
url=self.endpoints["getDetailWidgets"],
|
|
params={
|
|
"titleID": self.title,
|
|
"isTvodOnRow": "1",
|
|
"widgets": f'[{{"widgetType":"EpisodeList","widgetToken":"{token}"}}]'
|
|
},
|
|
headers={"Accept": "application/json"}
|
|
).json()
|
|
|
|
episodeListWidget = res_page["widgets"].get("episodeList", {})
|
|
for item in episodeListWidget.get("episodes", []):
|
|
ep_num = int(item.get("self", {}).get("sequenceNumber", 0))
|
|
episodes_list.append(Episode(
|
|
id_=item["detail"]["catalogId"],
|
|
title=show_name,
|
|
name=item["detail"]["title"],
|
|
season=product_details_season["seasonNumber"],
|
|
number=ep_num,
|
|
service=self.__class__,
|
|
data=item
|
|
))
|
|
|
|
pagination_data = res_page["widgets"].get("episodeList", {}).get("actions", {}).get("pagination", [])
|
|
token = next((quote(item.get("token")) for item in pagination_data if item.get("tokenType") == "NextPage"), None)
|
|
|
|
return Series(episodes_list)
|
|
|
|
def _discover_seasons_via_spa(self) -> list:
|
|
try:
|
|
headers = {
|
|
"accept": "application/json",
|
|
"device-memory": "8", "downlink": "10", "dpr": "2", "ect": "4g", "rtt": "50",
|
|
"viewport-width": "604", "x-amzn-client-ttl-seconds": "58.999",
|
|
"x-purpose": "navigation", "x-requested-with": "WebSPA",
|
|
}
|
|
base = self.region["base"]
|
|
if self.pv:
|
|
res_spa = self.session.get(f"https://{base}/detail/{self.title}", headers=headers)
|
|
url = (
|
|
f"https://{base}{res_spa.json()['redirect']}"
|
|
if "redirect" in res_spa.text else res_spa.url
|
|
)
|
|
else:
|
|
url = f"https://{base}/dp/{self.title}"
|
|
|
|
headers["referer"] = url
|
|
response = self.session.get(
|
|
url=url, params={"dvWebSPAClientVersion": "1.0.106799.0"}, headers=headers
|
|
)
|
|
if response.status_code != 200 or not response.text:
|
|
return []
|
|
|
|
spa_data = response.json()
|
|
seasons_data = None
|
|
for page in spa_data.get("page", []) or []:
|
|
assembly = page.get("assembly") or {}
|
|
for body in assembly.get("body", []) or []:
|
|
seasons_data = body.get("props", {}).get("atf", {}).get("state", {}).get("seasons")
|
|
if seasons_data:
|
|
break
|
|
if seasons_data:
|
|
break
|
|
|
|
seasons: list = []
|
|
if seasons_data:
|
|
first_group = next(iter(seasons_data.values()), [])
|
|
for season in first_group or []:
|
|
match = re.search(r"/detail/([A-Z0-9]{10,})/", season.get("seasonLink", ""))
|
|
if match and match.group(1) not in seasons:
|
|
seasons.append(match.group(1))
|
|
if seasons:
|
|
self.log.info(f" + Discovered {len(seasons)} season(s) via SPA detail page")
|
|
return seasons
|
|
except Exception as e:
|
|
self.log.debug(f"SPA season discovery failed: {e}")
|
|
return []
|
|
|
|
def _codec_fallback_chain(self, codec: str) -> list:
|
|
order = ["AV1", "H265", "H264"]
|
|
if codec not in order:
|
|
return [codec]
|
|
return order[order.index(codec):]
|
|
|
|
def _sync_vcodec_filter(self, codec: str) -> None:
|
|
enum_map = {"AV1": Video.Codec.AV1, "H265": Video.Codec.HEVC, "H264": Video.Codec.AVC}
|
|
target = enum_map.get(codec)
|
|
if not target:
|
|
return
|
|
params = getattr(getattr(self.ctx, "parent", None), "params", None)
|
|
if not isinstance(params, dict):
|
|
return
|
|
vlist = params.get("vcodec")
|
|
if isinstance(vlist, list) and vlist and target not in vlist:
|
|
vlist.append(target)
|
|
|
|
def _bitrate_for_codec(self, codec: str) -> str:
|
|
if self.bitrate_source == ParameterSource.COMMANDLINE:
|
|
return self.orig_bitrate
|
|
if self.manifest_type == "ISM":
|
|
return "CBR"
|
|
if self.vquality == "UHD" and self.range != "SDR":
|
|
return "CBR"
|
|
if codec == "AV1":
|
|
return "CVBR"
|
|
if codec == "H265" and self.range == "SDR":
|
|
return "CVBR+CBR"
|
|
return self.orig_bitrate
|
|
|
|
def _bitrate_candidates(self, codec: str) -> list:
|
|
if self.bitrate_source == ParameterSource.COMMANDLINE:
|
|
return [self.orig_bitrate]
|
|
cands = [self._bitrate_for_codec(codec)]
|
|
for b in ("CBR", "CVBR"):
|
|
if b not in cands:
|
|
cands.append(b)
|
|
return cands
|
|
|
|
def get_tracks(self, title: Title_T) -> Tracks:
|
|
if self.chapters_only:
|
|
return Tracks([])
|
|
self._check_entitlement(title)
|
|
self._warn_cdm_quality_mismatch()
|
|
|
|
is_hybrid = self.range == "HYBRID"
|
|
effective_range = "DV" if is_hybrid else self.range
|
|
video_protocol = "SmoothStreaming" if self.manifest_type == "ISM" else "DASH"
|
|
|
|
def _fetch_for_codec(codec: str) -> dict:
|
|
last: dict = {}
|
|
for bmode in self._bitrate_candidates(codec):
|
|
m = self.get_manifest(
|
|
title, video_codec=codec, bitrate_mode=bmode, quality=self.vquality,
|
|
hdr=effective_range, ignore_errors=True, protocol=video_protocol,
|
|
use_playlisted=self.playlisted,
|
|
)
|
|
if (self.range == "DV" or is_hybrid) and not m.get("vodPlaybackUrls"):
|
|
m = self.get_manifest(
|
|
title, video_codec=codec, bitrate_mode=bmode, quality=self.vquality,
|
|
hdr="HDR10", ignore_errors=True, protocol=video_protocol,
|
|
use_playlisted=self.playlisted,
|
|
)
|
|
if not self._usable_manifest(m) and self.device_token and not self.playlisted:
|
|
fb = self.get_manifest(
|
|
title, video_codec=codec, bitrate_mode=bmode, quality=self.vquality,
|
|
hdr=effective_range, ignore_errors=True, protocol=video_protocol,
|
|
use_playlisted=True,
|
|
)
|
|
if self._usable_manifest(fb):
|
|
m = fb
|
|
if self._usable_manifest(m):
|
|
self.bitrate = bmode
|
|
return m
|
|
last = m
|
|
return last
|
|
requested_codec = self.requested_vcodec
|
|
self.vcodec = requested_codec
|
|
self.bitrate = self._bitrate_for_codec(requested_codec)
|
|
codec_chain = self._codec_fallback_chain(requested_codec)
|
|
effective_vcodec = requested_codec
|
|
|
|
def _run_chain() -> dict:
|
|
nonlocal effective_vcodec
|
|
mani: dict = {}
|
|
for codec in codec_chain:
|
|
if len(codec_chain) > 1:
|
|
self.log.info(f" + Requesting {codec} video manifest...")
|
|
mani = _fetch_for_codec(codec)
|
|
if self._usable_manifest(mani):
|
|
if codec != requested_codec:
|
|
self.log.warning(f" - {requested_codec} unavailable for this title; using {codec}.")
|
|
effective_vcodec = codec
|
|
self.vcodec = codec
|
|
self._sync_vcodec_filter(codec)
|
|
return mani
|
|
if len(codec_chain) > 1:
|
|
self.log.warning(f" - {codec} not available for this title.")
|
|
return mani
|
|
|
|
manifest = _run_chain()
|
|
|
|
if not self._usable_manifest(manifest):
|
|
self.log.error(
|
|
f" - No usable manifest for this title with any codec ({', '.join(codec_chain)}). "
|
|
"Re-run with -d/--debug to see the Amazon error."
|
|
)
|
|
raise SystemExit(1)
|
|
|
|
if "rightsException" in manifest.get("returnedTitleRendition", {}).get("selectedEntitlement", {}):
|
|
self.log.error(" - The profile used does not have the rights to this title.")
|
|
return Tracks([])
|
|
|
|
self.session_handoff_token = (manifest.get("sessionization") or {}).get("sessionHandoffToken")
|
|
if isinstance(getattr(title, "data", None), dict):
|
|
title.data["_amzn_manifest"] = manifest
|
|
|
|
chosen_manifest = self.choose_manifest(manifest, self.cdn)
|
|
if not chosen_manifest:
|
|
self.log.error("No manifests available"); raise SystemExit(1)
|
|
|
|
streamingProtocol = (
|
|
manifest.get("vodPlaybackUrls", {}).get("result", {}).get("playbackUrls", {})
|
|
.get("urlMetadata", {}).get("streamingProtocol", "DASH")
|
|
)
|
|
if self.event:
|
|
devicetype = self.device.get("device_type")
|
|
manifest_url = f"{chosen_manifest['url']}?amznDtid={devicetype}&encoding=segmentBase"
|
|
elif streamingProtocol == "DASH":
|
|
manifest_url = self.clean_mpd_url(chosen_manifest["url"], False)
|
|
else:
|
|
manifest_url = chosen_manifest["url"]
|
|
|
|
self.log.info(f" + Manifest: {manifest_url}")
|
|
|
|
if streamingProtocol == "DASH":
|
|
_mpd_raw = self.session.get(manifest_url).text
|
|
_lang_order_map = _build_ordered_lang_map_from_mpd(_mpd_raw)
|
|
self.log.info(f" + MPD language map: {sum(len(v) for v in _lang_order_map.values())} representations indexed")
|
|
else:
|
|
_lang_order_map = {}
|
|
|
|
tracks = Tracks()
|
|
|
|
if streamingProtocol == "DASH":
|
|
tracks = Tracks([
|
|
x for x in iter(DASH.from_text(_mpd_raw, manifest_url).to_tracks(language="en"))
|
|
])
|
|
elif streamingProtocol == "SmoothStreaming":
|
|
_ism_tracks = Tracks()
|
|
for _t in iter(ISM.from_url(url=manifest_url, session=self.session).to_tracks(language="en")):
|
|
_ism_tracks.add(_t, warn_only=True)
|
|
tracks = _ism_tracks
|
|
else:
|
|
self.log.error(f"Unsupported manifest type: {streamingProtocol}"); raise SystemExit(1)
|
|
|
|
if _lang_order_map:
|
|
_apply_ordered_lang_map(tracks.audio, _lang_order_map)
|
|
self._tag_amzn_tracks(tracks, manifest)
|
|
|
|
need_separate_audio = ((self.aquality or self.vquality) != self.vquality
|
|
or self.amanifest == "CVBR" and (self.vcodec, self.bitrate) != ("H264", "CVBR")
|
|
or self.amanifest == "CBR" and (self.vcodec, self.bitrate) != ("H264", "CBR")
|
|
or self.amanifest == "H265" and self.vcodec != "H265"
|
|
or self.amanifest != "H265" and self.vcodec == "H265")
|
|
|
|
if not need_separate_audio:
|
|
audios = defaultdict(list)
|
|
for audio in tracks.audio:
|
|
audios[audio.language].append(audio)
|
|
for lang in audios:
|
|
if not any((x.bitrate or 0) >= 640000 for x in audios[lang]):
|
|
need_separate_audio = True
|
|
break
|
|
|
|
if need_separate_audio:
|
|
manifest_type = self.amanifest
|
|
self.log.info(f"Getting audio from {manifest_type} manifest for potential higher bitrate or better codec")
|
|
|
|
audio_manifest = self.get_manifest(
|
|
title=title,
|
|
video_codec="H264",
|
|
bitrate_mode="CVBR",
|
|
quality="HD",
|
|
hdr=None,
|
|
ignore_errors=True
|
|
)
|
|
|
|
if not audio_manifest:
|
|
self.log.warning(f" - Unable to get {manifest_type} audio manifests, skipping")
|
|
elif not (chosen_audio_manifest := self.choose_manifest(audio_manifest, self.cdn)):
|
|
self.log.warning(f" - No {manifest_type} audio manifests available, skipping")
|
|
else:
|
|
audio_mpd_url = self.clean_mpd_url(chosen_audio_manifest["url"], optimise=False)
|
|
if self.event:
|
|
devicetype = self.device.get("device_type")
|
|
audio_mpd_url = f"{chosen_audio_manifest['url']}?amznDtid={devicetype}&encoding=segmentBase"
|
|
|
|
self.log.info(" + Downloading Audio Manifest")
|
|
try:
|
|
audio_protocol = audio_manifest["vodPlaybackUrls"]["result"]["playbackUrls"]["urlMetadata"]["streamingProtocol"]
|
|
if audio_protocol == "DASH":
|
|
_a_raw = self.session.get(audio_mpd_url).text
|
|
_a_lang_order = _build_ordered_lang_map_from_mpd(_a_raw)
|
|
self.log.info(f" + Audio MPD language map: {sum(len(v) for v in _a_lang_order.values())} entries")
|
|
audio_tracks = DASH.from_text(_a_raw, audio_mpd_url).to_tracks(language="en")
|
|
if _a_lang_order:
|
|
_apply_ordered_lang_map(audio_tracks.audio, _a_lang_order)
|
|
elif audio_protocol == "SmoothStreaming":
|
|
audio_tracks = ISM.from_url(url=audio_mpd_url, session=self.session).to_tracks(language="en")
|
|
else:
|
|
audio_tracks = Tracks([])
|
|
|
|
self._tag_amzn_tracks(audio_tracks, audio_manifest)
|
|
tracks.add(audio_tracks.audio, warn_only=True)
|
|
except Exception as e:
|
|
self.log.warning(f" - Failed to parse audio manifest: {e}")
|
|
|
|
if (self.config.get("device") or {}).get(self.profile or "default", None):
|
|
self.log.info(" + Fetching DV/UHD manifest for Atmos audio (576kbps DD+)")
|
|
temp_device = self.device
|
|
temp_token = self.device_token
|
|
temp_id = self.device_id
|
|
|
|
if not self.device_token:
|
|
try:
|
|
self.register_device()
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
uhd_audio_manifest = self.get_manifest(
|
|
title=title,
|
|
video_codec="H265",
|
|
bitrate_mode="CVBR+CBR",
|
|
quality="UHD",
|
|
hdr="DV",
|
|
ignore_errors=True
|
|
)
|
|
except Exception:
|
|
uhd_audio_manifest = None
|
|
|
|
self.device = temp_device
|
|
self.device_token = temp_token
|
|
self.device_id = temp_id
|
|
|
|
if uhd_audio_manifest and (chosen_uhd := self.choose_manifest(uhd_audio_manifest, self.cdn)):
|
|
uhd_url = self.clean_mpd_url(chosen_uhd["url"], optimise=False)
|
|
self.log.info(" + Downloading DV/UHD audio manifest")
|
|
try:
|
|
uhd_prot = uhd_audio_manifest["vodPlaybackUrls"]["result"]["playbackUrls"]["urlMetadata"]["streamingProtocol"]
|
|
if uhd_prot == "DASH":
|
|
_uhd_raw = self.session.get(uhd_url).text
|
|
_uhd_lang_map = _build_ordered_lang_map_from_mpd(_uhd_raw)
|
|
uhd_tracks = DASH.from_text(_uhd_raw, uhd_url).to_tracks(language="en")
|
|
if _uhd_lang_map:
|
|
_apply_ordered_lang_map(uhd_tracks.audio, _uhd_lang_map)
|
|
self.log.info(f" + DV audio lang map: {sum(len(v) for v in _uhd_lang_map.values())} entries")
|
|
elif uhd_prot == "SmoothStreaming":
|
|
uhd_tracks = ISM.from_url(url=uhd_url, session=self.session).to_tracks(language="en")
|
|
else:
|
|
uhd_tracks = Tracks([])
|
|
self._tag_amzn_tracks(uhd_tracks, uhd_audio_manifest)
|
|
tracks.add(uhd_tracks.audio, warn_only=True)
|
|
_atmos_tracks = [x for x in uhd_tracks.audio if (x.bitrate or 0) >= 448000 and (x.channels or 0) >= 6]
|
|
if _atmos_tracks:
|
|
_best_kbps = max((x.bitrate or 0) for x in _atmos_tracks) // 1000
|
|
self.log.info(f" + Added {len(_atmos_tracks)} Atmos/high-bitrate audio track(s) from DV manifest (Best: {_best_kbps} kb/s)")
|
|
else:
|
|
self.log.info(" + DV audio manifest fetched (No Atmos found for this title)")
|
|
except Exception as e:
|
|
self.log.warning(f" - Failed to parse DV audio manifest: {e}")
|
|
else:
|
|
self.log.warning(" - DV/UHD audio manifest unavailable for this title")
|
|
|
|
self._post_process_audio(tracks.audio)
|
|
|
|
_all_codecs = list({v.codec for v in tracks.videos if v.codec})
|
|
if self.vcodec == "AV1":
|
|
filtered = [v for v in tracks.videos if v.codec and ("av01" in v.codec.lower() or "av1" in v.codec.lower())]
|
|
if not filtered:
|
|
self.log.error(
|
|
f" - No AV1 tracks found (found codecs: {_all_codecs}). "
|
|
"The manifest may not have AV1 or the codec string is unexpected."
|
|
)
|
|
raise SystemExit(1)
|
|
tracks.videos = filtered
|
|
elif self.vcodec == "H265":
|
|
filtered = [v for v in tracks.videos if v.codec and any(x in v.codec.lower() for x in ["hev1", "hvc1", "h265"])]
|
|
if filtered:
|
|
tracks.videos = filtered
|
|
else:
|
|
filtered = [v for v in tracks.videos if v.codec and "avc" in v.codec.lower()]
|
|
if filtered:
|
|
tracks.videos = filtered
|
|
|
|
actual_range = (
|
|
manifest.get("vodPlaybackUrls", {}).get("result", {}).get("playbackUrls", {})
|
|
.get("urlMetadata", {}).get("dynamicRange", "None")
|
|
)
|
|
for video in tracks.videos:
|
|
video.hdr10 = actual_range == "Hdr10"
|
|
video.dv = actual_range == "DolbyVision"
|
|
|
|
if (self.range == "DV" or is_hybrid) and actual_range != "DolbyVision":
|
|
friendly = {"Hdr10": "HDR10", "None": "SDR"}.get(actual_range, actual_range)
|
|
self.log.warning(f" - Dolby Vision not available for this title/region. Server returned: {friendly}")
|
|
|
|
if (self.range == "DV" or is_hybrid) and actual_range == "DolbyVision":
|
|
self.log.info(" + Hybrid mode: fetching HDR10 manifest for base layer...")
|
|
hdr10_manifest = self.get_manifest(
|
|
title=title,
|
|
video_codec=effective_vcodec,
|
|
bitrate_mode=self.bitrate,
|
|
quality=self.vquality,
|
|
hdr="HDR10",
|
|
ignore_errors=True
|
|
)
|
|
if hdr10_manifest and hdr10_manifest.get("vodPlaybackUrls"):
|
|
chosen_hdr10 = self.choose_manifest(hdr10_manifest, self.cdn)
|
|
if chosen_hdr10:
|
|
hdr10_url = self.clean_mpd_url(chosen_hdr10["url"], False)
|
|
self.log.info(f" + HDR10 Manifest (base layer): {hdr10_url}")
|
|
try:
|
|
hdr10_protocol = hdr10_manifest["vodPlaybackUrls"]["result"]["playbackUrls"]["urlMetadata"]["streamingProtocol"]
|
|
if hdr10_protocol == "DASH":
|
|
hdr10_tracks = DASH.from_url(url=hdr10_url, session=self.session).to_tracks(language="en")
|
|
elif hdr10_protocol == "SmoothStreaming":
|
|
hdr10_tracks = ISM.from_url(url=hdr10_url, session=self.session).to_tracks(language="en")
|
|
else:
|
|
hdr10_tracks = Tracks([])
|
|
for video in hdr10_tracks.videos:
|
|
video.hdr10 = True
|
|
video.dv = False
|
|
hdr10_video_count = len(list(hdr10_tracks.videos))
|
|
tracks.add(hdr10_tracks.videos, warn_only=True)
|
|
self.log.info(f" + Added {hdr10_video_count} HDR10 base-layer video track(s) for hybrid mux")
|
|
except Exception as e:
|
|
self.log.warning(f" - Failed to fetch HDR10 base layer for hybrid: {e}")
|
|
else:
|
|
self.log.warning(" - No HDR10 manifest CDN available for hybrid mode")
|
|
else:
|
|
self.log.warning(" - HDR10 manifest unavailable for hybrid and DV-only tracks will be used")
|
|
|
|
for sub in manifest.get("timedTextUrls", {}).get("result", {}).get("subtitleUrls", []) + \
|
|
manifest.get("timedTextUrls", {}).get("result", {}).get("forcedNarrativeUrls", []):
|
|
url = sub["url"]
|
|
url_ext = os.path.splitext(url.split("?")[0])[1].lstrip(".").lower()
|
|
codec_map = {
|
|
"ttml": Subtitle.Codec.TimedTextMarkupLang,
|
|
"dfxp": Subtitle.Codec.TimedTextMarkupLang,
|
|
"vtt": Subtitle.Codec.WebVTT,
|
|
"srt": Subtitle.Codec.SubRip,
|
|
}
|
|
detected_codec = codec_map.get(url_ext, Subtitle.Codec.TimedTextMarkupLang)
|
|
|
|
tracks.add(Subtitle(
|
|
id_=f"{sub['trackGroupId']}_{sub['languageCode']}_{sub['type']}_{sub['subtype']}",
|
|
url=url,
|
|
codec=detected_codec,
|
|
language=_resolve_subtitle_language(sub["languageCode"], url),
|
|
forced="ForcedNarrative" in sub["type"],
|
|
sdh=sub["type"].lower() == "sdh"
|
|
), warn_only=True)
|
|
|
|
try:
|
|
images = title.data.get("detail", {}).get("images", {})
|
|
if isinstance(title, Movie):
|
|
image_url = images.get("packshot") or images.get("titleshot") or images.get("covershot")
|
|
else:
|
|
image_url = images.get("covershot") or images.get("packshot") or images.get("titleshot")
|
|
if image_url and image_url.strip():
|
|
tracks.add(Attachment.from_url(
|
|
url=image_url.strip(),
|
|
name="cover",
|
|
mime_type="image/jpeg",
|
|
session=self.session,
|
|
))
|
|
except Exception as e:
|
|
self.log.warning(f" - Attachment failed: {e}")
|
|
if not self.no_true_region and self.vquality != "UHD" and tracks.videos:
|
|
self.manage_session(tracks.videos[0])
|
|
|
|
return tracks
|
|
|
|
def get_chapters(self, title: Title_T) -> Chapters:
|
|
manifest = title.data.get("_amzn_manifest") if isinstance(getattr(title, "data", None), dict) else None
|
|
if not manifest:
|
|
manifest = self.get_manifest(
|
|
title,
|
|
video_codec=self.vcodec,
|
|
bitrate_mode=self.bitrate,
|
|
quality=self.vquality,
|
|
hdr=self.range,
|
|
ignore_errors=True
|
|
)
|
|
if not manifest:
|
|
return Chapters()
|
|
|
|
chapters = self._chapters_from_transitions(manifest)
|
|
if chapters:
|
|
self.log.info(f" + Found {len(chapters)} chapter marker(s) from transition timecodes")
|
|
return Chapters(chapters)
|
|
|
|
return Chapters(self._chapters_from_xray(manifest))
|
|
|
|
def _chapters_from_transitions(self, manifest: dict) -> list:
|
|
result = manifest.get("transitionTimecodes", {}).get("result", {})
|
|
events = result.get("events")
|
|
if not events:
|
|
return []
|
|
|
|
def fmt(ms):
|
|
s = ms / 1000
|
|
return f"{int(s // 3600):02d}:{int((s % 3600) // 60):02d}:{s % 60:06.3f}"
|
|
|
|
chapters = []
|
|
for event in sorted(events, key=lambda e: e.get("startTimeMs", 0)):
|
|
start_ms = event.get("startTimeMs")
|
|
if start_ms is None:
|
|
continue
|
|
event_type = event.get("eventType", "")
|
|
if event_type == "SKIP_INTRO":
|
|
chapters.append(Chapter(name="Intro", timestamp=fmt(start_ms)))
|
|
if event.get("endTimeMs"):
|
|
chapters.append(Chapter(name="Main Content", timestamp=fmt(event["endTimeMs"])))
|
|
elif event_type == "SKIP_RECAP":
|
|
chapters.append(Chapter(name="Recap", timestamp=fmt(start_ms)))
|
|
if event.get("endTimeMs"):
|
|
chapters.append(Chapter(name="Episode Start", timestamp=fmt(event["endTimeMs"])))
|
|
elif event_type == "END_CREDITS":
|
|
chapters.append(Chapter(name="Credits", timestamp=fmt(start_ms)))
|
|
return chapters
|
|
|
|
def _chapters_from_xray(self, manifest: dict) -> list:
|
|
if "vodXrayMetadata" not in manifest or "error" in manifest["vodXrayMetadata"]:
|
|
return []
|
|
try:
|
|
xray_params = {
|
|
"pageId": "fullScreen",
|
|
"pageType": "xray",
|
|
"serviceToken": json.dumps({
|
|
"consumptionType": "Streaming",
|
|
"deviceClass": "normal",
|
|
"playbackMode": "playback",
|
|
"vcid": json.loads(manifest["vodXrayMetadata"]["result"]["parameters"]["serviceToken"])["vcid"]
|
|
}),
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.config["device_types"]["browser"],
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"gascEnabled": str(self.pv).lower(),
|
|
"decorationScheme": "none",
|
|
"version": "inception-v2",
|
|
"uxLocale": "en-US",
|
|
"featureScheme": "XRAY_WEB_2020_V1"
|
|
}
|
|
xray = self.session.get(url=self.endpoints["xray"], params=xray_params).json().get("page")
|
|
except Exception:
|
|
return []
|
|
|
|
if not xray:
|
|
return []
|
|
|
|
try:
|
|
widgets = xray["sections"]["center"]["widgets"]["widgetList"]
|
|
scenes = next((x for x in widgets if x["tabType"] == "scenesTab"), None)
|
|
if not scenes:
|
|
return []
|
|
scenes = scenes["widgets"]["widgetList"][0]["items"]["itemList"]
|
|
except (KeyError, IndexError):
|
|
return []
|
|
|
|
chapters = []
|
|
for scene in scenes:
|
|
chapter_title = scene["textMap"]["PRIMARY"]
|
|
match = re.search(r"(\d+\. |)(.+)", chapter_title)
|
|
if match:
|
|
chapter_title = match.group(2)
|
|
timecode = scene["textMap"]["TERTIARY"].replace("Starts at ", "")
|
|
chapters.append(Chapter(name=chapter_title, timestamp=timecode))
|
|
return chapters
|
|
|
|
def _check_entitlement(self, title) -> None:
|
|
try:
|
|
base = self.region.get("base_manifest")
|
|
if not base:
|
|
return
|
|
params = {
|
|
"itemId": title.id,
|
|
"presentationScheme": "android-tv-react",
|
|
"deviceTypeID": self.device.get("device_type"),
|
|
"deviceID": self.device_id,
|
|
}
|
|
headers = {"Accept": "application/json"}
|
|
_bearer = self.actor_token or self.device_token
|
|
if _bearer:
|
|
params["roles"] = "playback-envelope-supported"
|
|
headers["Authorization"] = f"Bearer {_bearer}"
|
|
else:
|
|
params["firmware"] = ""
|
|
params["roles"] = "prime-offer-supported,svod-supported"
|
|
params["clientFeatures"] = "EnableBuyBoxV2"
|
|
|
|
res = self.session.get(
|
|
url=f"https://{base}/lrcedge/getDataByJavaTransform/v1/lr/detailsPage/detailsPageATF",
|
|
params=params,
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
if res.status_code != 200:
|
|
self.log.debug(f"Entitlement pre-check unavailable ({res.status_code})")
|
|
return
|
|
|
|
resource = (res.json() or {}).get("resource", {}) or {}
|
|
message = (
|
|
resource.get("entitlementMessaging", {})
|
|
.get("ENTITLEMENT_MESSAGE_SLOT_DETAIL", {})
|
|
.get("message")
|
|
)
|
|
if message:
|
|
if any(k in message.lower() for k in ("join", "trial", "subscribe", "rent", "buy", "add-on", "add on")):
|
|
self.log.warning(f" - Entitlement: {message} (this title may need a purchase/add-on)")
|
|
else:
|
|
self.log.info(f" + Entitlement: {message}")
|
|
|
|
apply_hdr = resource.get("applyHdr")
|
|
apply_uhd = resource.get("applyUhd")
|
|
bits = []
|
|
if apply_uhd is not None:
|
|
bits.append(f"UHD {'available' if apply_uhd else 'not available'}")
|
|
if apply_hdr is not None:
|
|
bits.append(f"HDR {'available' if apply_hdr else 'not available'}")
|
|
if bits:
|
|
self.log.info(f" + Amazon advertises: {', '.join(bits)} for this title")
|
|
except Exception as e:
|
|
self.log.debug(f"Entitlement pre-check failed: {e}")
|
|
|
|
def _bearer(self) -> Optional[str]:
|
|
if self.living_room and self.actor_token:
|
|
return self.actor_token
|
|
return self.device_token
|
|
|
|
def _get_primary_profile(self) -> Optional[str]:
|
|
try:
|
|
base = self.region.get("base_manifest")
|
|
if not base or not self.device_token:
|
|
return None
|
|
res = self.session.get(
|
|
url=f"https://{base}/lrcedge/getDataByJavaTransform/v1/lr/profiles/profileSelection",
|
|
params={"deviceTypeID": self.device.get("device_type"), "deviceID": self.device_id},
|
|
headers={"Authorization": f"Bearer {self.device_token}", "Accept": "application/json"},
|
|
timeout=15,
|
|
).json()
|
|
profiles = (res.get("resource") or {}).get("profiles") or []
|
|
default = next((p for p in profiles if p.get("isDefaultProfile")), None) or (profiles[0] if profiles else None)
|
|
return default.get("profileId") if default else None
|
|
except Exception as e:
|
|
self.log.debug(f"Profile lookup failed: {e}")
|
|
return None
|
|
|
|
def _ensure_actor_token(self) -> Optional[str]:
|
|
if self.actor_token:
|
|
return self.actor_token
|
|
if not self.device_token or not self.device_refresh_token:
|
|
return None
|
|
|
|
_profile = self.profile or "default"
|
|
cache = Cacher("AMZN").get(f"actor_token_{_profile}")
|
|
if cache and cache.data and cache.data.get("token") and cache.data.get("expires_in", 0) > int(time.time()):
|
|
self.actor_token = cache.data["token"]
|
|
self.profile_id = cache.data.get("profile_id")
|
|
self.log.debug(" + Using cached actor token")
|
|
return self.actor_token
|
|
|
|
try:
|
|
profile_id = self._get_primary_profile()
|
|
if not profile_id:
|
|
self.log.debug(" - No primary profile found for actor token")
|
|
return None
|
|
res = self.session.post(
|
|
url=self.endpoints["token"],
|
|
headers={"Content-Type": "application/json"},
|
|
json={
|
|
"actor_id": profile_id,
|
|
"app_name": "AIV",
|
|
"requested_token_type": "actor_access_token",
|
|
"source_token_type": "refresh_token",
|
|
"source_device_tokens": [{
|
|
"device_type": self.device.get("device_type"),
|
|
"account_refresh_token": {"token": self.device_refresh_token},
|
|
}],
|
|
},
|
|
).json()
|
|
device_tokens = res.get("device_tokens") or []
|
|
token = (device_tokens[0].get("actor_access_token") or {}).get("token") if device_tokens else None
|
|
if not token:
|
|
self.log.debug(f" - Actor token exchange returned no token (Keys: {list(res.keys())})")
|
|
return None
|
|
self.actor_token = token
|
|
self.profile_id = profile_id
|
|
cache.set(
|
|
{"token": token, "profile_id": profile_id, "expires_in": int(time.time()) + 3000},
|
|
int(time.time()) + 3000,
|
|
)
|
|
self.log.info(" + Acquired LivingRoomPlayer actor token")
|
|
return token
|
|
except Exception as e:
|
|
self.log.debug(f"Actor token acquisition failed: {e}")
|
|
return None
|
|
|
|
def _livingroom_envelope(self, title) -> Optional[str]:
|
|
token = self._ensure_actor_token()
|
|
if not token:
|
|
return None
|
|
try:
|
|
base = self.region.get("base_manifest")
|
|
res = self.session.get(
|
|
url=f"https://{base}/lrcedge/getDataByJavaTransform/v1/lr/detailsPage/detailsPageATF",
|
|
params={
|
|
"itemId": title.id,
|
|
"presentationScheme": "android-tv-react",
|
|
"deviceTypeID": self.device.get("device_type"),
|
|
"deviceID": self.device_id,
|
|
"roles": "playback-envelope-supported",
|
|
},
|
|
headers={"Authorization": f"Bearer {token}", "Accept": "application/json"},
|
|
timeout=20,
|
|
).json()
|
|
resource = (res or {}).get("resource") or {}
|
|
for action in resource.get("actions", []) or []:
|
|
pem = (action.get("metadata") or {}).get("playbackExperienceMetadata") or {}
|
|
if pem.get("playbackEnvelope"):
|
|
self.playbackInfo = {"titleID": title.id, "playbackExperienceMetadata": pem}
|
|
return pem["playbackEnvelope"]
|
|
self.log.debug(" - LivingRoom detailsPage returned no playback envelope")
|
|
except Exception as e:
|
|
self.log.debug(f"LivingRoom envelope fetch failed: {e}")
|
|
return None
|
|
|
|
def playbackEnvelope_data(self, title):
|
|
try:
|
|
res = self.session.get(
|
|
url=self.endpoints["metadata"],
|
|
params={
|
|
"metadataToEnrich": json.dumps({"placement": "HOVER", "playback": "true", "preroll": "true", "trailer": "true", "watchlist": "true"}),
|
|
"titleIDsToEnrich": json.dumps([title.id]),
|
|
"currentUrl": f"https://{self.region['base']}/"
|
|
},
|
|
headers={
|
|
"device-memory": "8",
|
|
"x-amzn-requestid": "".join(secrets.choice(string.ascii_uppercase + string.digits) for _ in range(20)),
|
|
"x-requested-with": "XMLHttpRequest"
|
|
}
|
|
)
|
|
|
|
if res.status_code != 200:
|
|
self.log.error(f"Unable to get playbackEnvelope: {res.text}"); raise SystemExit(1)
|
|
|
|
data = res.json()
|
|
enrichment = data.get("enrichments", {}).get(title.id, {})
|
|
playback_actions = [
|
|
a for a in enrichment.get("playbackActions", [])
|
|
if a.get("playbackExperienceMetadata", {}).get("playbackEnvelope")
|
|
]
|
|
if playback_actions:
|
|
pem = playback_actions[0]["playbackExperienceMetadata"]
|
|
self.playbackInfo = {"titleID": title.id, "playbackExperienceMetadata": pem}
|
|
return pem["playbackEnvelope"]
|
|
|
|
preroll = enrichment.get("prerollsEnvelope") or {}
|
|
preroll_env = preroll.get("playbackEnvelope")
|
|
if preroll_env:
|
|
self.playbackInfo = {"titleID": title.id, "playbackExperienceMetadata": preroll}
|
|
return preroll_env
|
|
focus = enrichment.get("entitlementCues", {}).get("focusMessage", {}).get("message", "")
|
|
if focus and any(k in focus.lower() for k in ("trial", "rent", "buy", "subscribe", "join")):
|
|
self.log.error(
|
|
f" - This account is not entitled to '{title}'. Amazon returned: \"{focus}\". "
|
|
"Account needs a purchase/rental/subscription, or the cookies belong to "
|
|
"a different account."
|
|
); raise SystemExit(1)
|
|
self.log.error(
|
|
f" - No playback envelope returned for '{title}' "
|
|
f"(not entitled, or bad cookies). Entitlement cue: {focus or 'none'}"
|
|
); raise SystemExit(1)
|
|
except SystemExit:
|
|
raise
|
|
except Exception as e:
|
|
self.log.error(f"Unable to get playbackEnvelope: {e}"); raise SystemExit(1)
|
|
|
|
def playbackEnvelope_update(self, playbackInfo):
|
|
if not playbackInfo:
|
|
return playbackInfo
|
|
try:
|
|
pem = playbackInfo.get("playbackExperienceMetadata", {}) or {}
|
|
expiry = pem.get("expiryTime", 0)
|
|
if expiry and (int(expiry) / 1000) > time.time():
|
|
return playbackInfo
|
|
|
|
correlation_id = pem.get("correlationId")
|
|
title_id = playbackInfo.get("titleID")
|
|
if not correlation_id or not title_id:
|
|
return playbackInfo
|
|
|
|
self.log.debug(" + Refreshing expired playback envelope")
|
|
res = self.session.post(
|
|
url=self.endpoints["refreshplayback"],
|
|
params={
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.device["device_type"],
|
|
"gascEnabled": str(self.pv).lower(),
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"uxLocale": "en_EN",
|
|
"firmware": "1",
|
|
"version": "1",
|
|
"nerid": self.generate_nerid(),
|
|
},
|
|
data=json.dumps({
|
|
"deviceId": self.device_id,
|
|
"deviceTypeId": self.device["device_type"],
|
|
"identifiers": {title_id: correlation_id},
|
|
"geoToken": "null",
|
|
"identityContext": "null",
|
|
}),
|
|
)
|
|
if res.status_code != 200:
|
|
return playbackInfo
|
|
response_data = res.json().get("response", {})
|
|
if not isinstance(response_data, dict):
|
|
return playbackInfo
|
|
new_pe = response_data.get(title_id, {}).get("playbackExperience")
|
|
if not new_pe:
|
|
return playbackInfo
|
|
new_expiry = new_pe.get("expiryTime")
|
|
if new_expiry:
|
|
new_pe["expiryTime"] = int(new_expiry * 1000)
|
|
refreshed = {"titleID": title_id, "playbackExperienceMetadata": new_pe}
|
|
self.playbackInfo = refreshed
|
|
if new_pe.get("playbackEnvelope"):
|
|
self.playbackEnvelope = new_pe["playbackEnvelope"]
|
|
return refreshed
|
|
except Exception as e:
|
|
self.log.debug(f"Playback envelope refresh failed (non-fatal): {e}")
|
|
return playbackInfo
|
|
|
|
def _technologies(self, protocol: str) -> list:
|
|
if protocol == "SmoothStreaming":
|
|
return ["SmoothStreaming"]
|
|
if self.manifest_type_source == ParameterSource.COMMANDLINE and self.manifest_type == "DASH":
|
|
return ["DASH"]
|
|
return ["DASH", "SmoothStreaming"]
|
|
|
|
def _build_manifest_payload(self, title, video_codec, bitrate_mode, quality, hdr, protocol,
|
|
use_playlisted):
|
|
bitrate_adaptations = ["CVBR", "CBR"] if bitrate_mode in ("CVBR+CBR", "CVBR,CBR") else [bitrate_mode]
|
|
range_fmt = self.VIDEO_RANGE_MAP.get(hdr, "None")
|
|
drm_type = "PlayReady" if self.playready else "Widevine"
|
|
|
|
if use_playlisted and self.device_token:
|
|
def build_tech():
|
|
return {
|
|
"bitrateAdaptations": bitrate_adaptations,
|
|
"codecs": [video_codec],
|
|
"drmType": drm_type,
|
|
"drmKeyScheme": "SingleKey" if self.playready else "DualKey",
|
|
"drmStrength": "L40",
|
|
"dynamicRangeFormats": [range_fmt],
|
|
"edgeDeliveryAuthorizationSchemes": ["PVExchangeV1", "Transparent"],
|
|
"fragmentRepresentations": ["ByteOffsetRange", "SeparateFile"],
|
|
"frameRates": ["Standard", "High"],
|
|
"segmentInfoType": "Base",
|
|
"stitchType": "MultiPeriod",
|
|
"timedTextRepresentations": ["BurnedIn", "NotInManifestNorStream", "SeparateStreamInManifest"],
|
|
"trickplayRepresentations": ["NotInManifestNorStream"],
|
|
"variableAspectRatio": "supported",
|
|
"vastTimelineType": "Absolute",
|
|
"manifestThinningToSupportedResolution": "Forbidden",
|
|
}
|
|
|
|
supported_techs = self._technologies(protocol)
|
|
techs = {t: build_tech() for t in supported_techs}
|
|
|
|
return {
|
|
"globalParameters": {
|
|
"deviceCapabilityFamily": "LivingRoomPlayer",
|
|
"playbackEnvelope": self.playbackEnvelope,
|
|
"capabilityDiscriminators": {"discriminators": {"software": {}, "version": 1}},
|
|
},
|
|
"timedTextUrlsRequest": {"supportedTimedTextFormats": ["TTMLv2", "DFXP"]},
|
|
"transitionTimecodesRequest": {},
|
|
"vodPlaylistedPlaybackUrlsRequest": {
|
|
"device": {
|
|
"displayBasedVending": "supported",
|
|
"displayHeight": 2304,
|
|
"displayWidth": 4096,
|
|
"hdcpLevel": "2.3",
|
|
"maxVideoResolution": "2160p",
|
|
"category": "Tv",
|
|
"platform": "Android",
|
|
"streamingTechnologies": techs,
|
|
"supportedStreamingTechnologies": supported_techs,
|
|
},
|
|
"playbackSettingsRequest": {
|
|
"firmware": "UNKNOWN",
|
|
"playerType": self.player,
|
|
"responseFormatVersion": "1.0.0",
|
|
"titleId": title.id,
|
|
},
|
|
},
|
|
"vodXrayMetadataRequest": {
|
|
"xrayDeviceClass": "normal",
|
|
"xrayPlaybackMode": "playback",
|
|
"xrayToken": "XRAY_REIGN_3PLR_2025_V2",
|
|
},
|
|
}
|
|
|
|
if not self.device_token:
|
|
if protocol == "SmoothStreaming":
|
|
self.log.warning(" - ISM/SmoothStreaming requires a registered device; using DASH for web playback.")
|
|
global_params = {
|
|
"deviceCapabilityFamily": "WebPlayer",
|
|
"playbackEnvelope": self.playbackEnvelope,
|
|
"capabilityDiscriminators": {
|
|
"operatingSystem": {"name": "Windows", "version": "10.0"},
|
|
"middleware": {"name": "EdgeNext", "version": "136.0.0.0"},
|
|
"nativeApplication": {"name": "EdgeNext", "version": "136.0.0.0"},
|
|
"hfrControlMode": "Legacy",
|
|
"displayResolution": {"height": 2304, "width": 4096}
|
|
}
|
|
}
|
|
audit_request = {}
|
|
vod_request = {
|
|
"device": {
|
|
"hdcpLevel": "2.2" if quality == "UHD" else "1.4",
|
|
"maxVideoResolution": ("1080p" if quality == "HD" else "480p" if quality == "SD" else "2160p"),
|
|
"supportedStreamingTechnologies": ["DASH"],
|
|
"streamingTechnologies": {
|
|
"DASH": {
|
|
"bitrateAdaptations": bitrate_adaptations,
|
|
"codecs": [video_codec],
|
|
"drmKeyScheme": "SingleKey" if self.playready else "DualKey",
|
|
"drmType": drm_type,
|
|
"dynamicRangeFormats": range_fmt,
|
|
"fragmentRepresentations": ["ByteOffsetRange", "SeparateFile"],
|
|
"frameRates": ["Standard"],
|
|
"stitchType": "MultiPeriod",
|
|
"segmentInfoType": "Base",
|
|
"timedTextRepresentations": ["NotInManifestNorStream", "SeparateStreamInManifest"],
|
|
"trickplayRepresentations": ["NotInManifestNorStream"],
|
|
"variableAspectRatio": "supported"
|
|
}
|
|
},
|
|
"displayWidth": 4096,
|
|
"displayHeight": 2304
|
|
},
|
|
"ads": {"sitePageUrl": "", "gdpr": {"enabled": "false", "consentMap": {}}},
|
|
"playbackCustomizations": {},
|
|
"playbackSettingsRequest": {
|
|
"firmware": "UNKNOWN",
|
|
"playerType": self.player,
|
|
"responseFormatVersion": "1.0.0",
|
|
"titleId": title.id
|
|
}
|
|
}
|
|
else:
|
|
global_params = {
|
|
"deviceCapabilityFamily": "AndroidPlayer",
|
|
"playbackEnvelope": self.playbackEnvelope,
|
|
"capabilityDiscriminators": {"discriminators": {"software": {}, "version": 1}}
|
|
}
|
|
audit_request = {"device": {"category": "Tv", "platform": "Android"}}
|
|
|
|
def _android_tech():
|
|
return {
|
|
"bitrateAdaptations": bitrate_adaptations,
|
|
"codecs": [video_codec],
|
|
"drmType": drm_type,
|
|
"drmKeyScheme": "SingleKey" if self.playready else "DualKey",
|
|
"drmStrength": "L40",
|
|
"dynamicRangeFormats": [range_fmt],
|
|
"edgeDeliveryAuthorizationSchemes": ["PVExchangeV1", "Transparent"],
|
|
"fragmentRepresentations": ["ByteOffsetRange", "SeparateFile"],
|
|
"frameRates": ["Standard", "High"],
|
|
"segmentInfoType": "Base",
|
|
"stitchType": "MultiPeriod",
|
|
"timedTextRepresentations": ["BurnedIn", "NotInManifestNorStream", "SeparateStreamInManifest"],
|
|
"trickplayRepresentations": ["NotInManifestNorStream"],
|
|
"variableAspectRatio": "supported",
|
|
"vastTimelineType": "Absolute",
|
|
"manifestThinningToSupportedResolution": "Forbidden"
|
|
}
|
|
|
|
technologies = self._technologies(protocol)
|
|
tech_block = {t: _android_tech() for t in technologies}
|
|
vod_request = {
|
|
"ads": {},
|
|
"device": {
|
|
"displayBasedVending": "supported",
|
|
"displayHeight": 2304,
|
|
"displayWidth": 4096,
|
|
"streamingTechnologies": tech_block,
|
|
"acceptedCreativeApis": [],
|
|
"category": "Tv",
|
|
"hdcpLevel": "2.2",
|
|
"maxVideoResolution": "2160p",
|
|
"platform": "Android",
|
|
"supportedStreamingTechnologies": technologies
|
|
},
|
|
"playbackCustomizations": {},
|
|
"playbackSettingsRequest": {
|
|
"firmware": "UNKNOWN",
|
|
"playerType": self.player,
|
|
"responseFormatVersion": "1.0.0",
|
|
"titleId": title.id
|
|
}
|
|
}
|
|
|
|
return {
|
|
"globalParameters": global_params,
|
|
"auditPingsRequest": audit_request,
|
|
"playbackDataRequest": {},
|
|
"timedTextUrlsRequest": {"supportedTimedTextFormats": ["TTMLv2", "DFXP"]},
|
|
"trickplayUrlsRequest": {},
|
|
"transitionTimecodesRequest": {},
|
|
"vodPlaybackUrlsRequest": vod_request,
|
|
"vodXrayMetadataRequest": {
|
|
"xrayDeviceClass": "normal",
|
|
"xrayPlaybackMode": "playback",
|
|
"xrayToken": "XRAY_WEB_2023_V2"
|
|
}
|
|
}
|
|
|
|
@staticmethod
|
|
def _normalize_playlisted_manifest(manifest: dict) -> dict:
|
|
playlisted = manifest.get("vodPlaylistedPlaybackUrls", {}) or {}
|
|
converted = dict(manifest)
|
|
|
|
if "error" in playlisted:
|
|
converted["vodPlaybackUrls"] = {"error": playlisted["error"]}
|
|
return converted
|
|
|
|
playback_urls = playlisted.get("result", {}).get("playbackUrls", {}) or {}
|
|
playlist = playback_urls.get("intraTitlePlaylist", []) or []
|
|
main = next((x for x in playlist if x.get("type") == "Main"), playlist[0] if playlist else None)
|
|
|
|
if not main:
|
|
converted["vodPlaybackUrls"] = {"result": {"playbackUrls": {"urlSets": [], "urlMetadata": {}}}}
|
|
return converted
|
|
|
|
url_sets = [
|
|
{"cdn": u.get("cdn"), "url": u.get("url"), "consumptionId": u.get("consumptionId")}
|
|
for u in main.get("urls", []) if u.get("url")
|
|
]
|
|
converted["vodPlaybackUrls"] = {
|
|
"result": {
|
|
"playbackUrls": {
|
|
"urlSets": url_sets,
|
|
"urlMetadata": main.get("manifestMetadata", {}),
|
|
}
|
|
}
|
|
}
|
|
return converted
|
|
|
|
def get_manifest(self, title, video_codec, bitrate_mode, quality, hdr, ignore_errors=False,
|
|
protocol="DASH", use_playlisted=False, retries=3):
|
|
use_playlisted = use_playlisted or self.living_room
|
|
if use_playlisted and not self.device_token:
|
|
self.log.warning(" - Playlisted (LivingRoomPlayer) path needs a registered device. Trying classic run.")
|
|
use_playlisted = False
|
|
|
|
for attempt in range(retries):
|
|
if attempt == 0 or not self.playbackInfo:
|
|
lr_env = self._livingroom_envelope(title) if self.living_room else None
|
|
if self.living_room and not lr_env:
|
|
self.log.warning(" - LivingRoom envelope unavailable. Trying classic run.")
|
|
self.living_room = False
|
|
use_playlisted = self.playlisted
|
|
self.playbackEnvelope = lr_env or self.playbackEnvelope_data(title)
|
|
else:
|
|
self.log.debug(f" + Manifest retry {attempt + 1}/{retries}; refreshing playback envelope")
|
|
try:
|
|
self.playbackInfo["playbackExperienceMetadata"]["expiryTime"] = 0
|
|
except Exception:
|
|
pass
|
|
self.playbackInfo = self.playbackEnvelope_update(self.playbackInfo)
|
|
self.playbackEnvelope = (
|
|
(self.playbackInfo or {}).get("playbackExperienceMetadata", {}).get("playbackEnvelope")
|
|
or self.playbackEnvelope
|
|
)
|
|
|
|
data_dict = self._build_manifest_payload(
|
|
title, video_codec, bitrate_mode, quality, hdr, protocol, use_playlisted,
|
|
)
|
|
|
|
res = self.session.post(
|
|
url=self.endpoints["playback"],
|
|
params={
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.device["device_type"],
|
|
"gascEnabled": str(self.pv).lower(),
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"uxLocale": "en_EN",
|
|
"firmware": 1,
|
|
"titleId": title.id,
|
|
"nerid": self.generate_nerid(),
|
|
},
|
|
data=json.dumps(data_dict),
|
|
headers={
|
|
"Authorization": f"Bearer {self._bearer()}" if self._bearer() else None,
|
|
},
|
|
)
|
|
|
|
try:
|
|
manifest = res.json()
|
|
except json.JSONDecodeError:
|
|
if attempt < retries - 1:
|
|
time.sleep(2 ** attempt)
|
|
continue
|
|
if ignore_errors:
|
|
return {}
|
|
self.log.error(f" - Amazon reported an error when obtaining the Playback Manifest\n{res.text}"); raise SystemExit(1)
|
|
if "vodPlaylistedPlaybackUrls" in manifest:
|
|
manifest = self._normalize_playlisted_manifest(manifest)
|
|
|
|
vod = manifest.get("vodPlaybackUrls", {})
|
|
|
|
if "error" in vod:
|
|
message = vod["error"].get("message", "unknown error")
|
|
if ignore_errors:
|
|
self.log.warning(f" - {video_codec} manifest error: {message}")
|
|
return {}
|
|
self.log.error(f" - Amazon reported an error when obtaining the Playback Manifest: {message}"); raise SystemExit(1)
|
|
for resource in ("PlaybackUrls", "AudioVideoUrls"):
|
|
err = manifest.get("errorsByResource", {}).get(resource)
|
|
if err and err.get("errorCode") not in (None, "PRS.NoRights.NotOwned"):
|
|
detail = f"{err.get('message')} [{err.get('errorCode')}]"
|
|
if ignore_errors:
|
|
self.log.warning(f" - {video_codec} {resource} error: {detail}")
|
|
return {}
|
|
self.log.error(f" - Amazon had an error with the {resource}: {detail}"); raise SystemExit(1)
|
|
|
|
return manifest
|
|
|
|
return {}
|
|
|
|
def get_widevine_service_certificate(self, **_: Any) -> str:
|
|
return self.config["certificate"]
|
|
|
|
def get_widevine_license(self, challenge: bytes, title: Title_T, track, **_) -> str:
|
|
return self._get_license(challenge, title, track, widevine=True)
|
|
|
|
def get_playready_license(self, challenge: bytes, title: Title_T, track, **_) -> str:
|
|
return self._get_license(challenge, title, track, widevine=False)
|
|
|
|
def _get_license(self, challenge: bytes, title: Title_T, track, widevine: bool):
|
|
if not self.device_token:
|
|
try:
|
|
self.register_device()
|
|
except Exception:
|
|
pass
|
|
|
|
challenge_bytes = challenge if isinstance(challenge, bytes) else challenge.encode("utf-8")
|
|
encoded_challenge = base64.b64encode(challenge_bytes).decode("utf-8")
|
|
|
|
amzn = track.data.get("amzn", {}) if isinstance(getattr(track, "data", None), dict) else {}
|
|
envelope = amzn.get("envelope") or self.playbackEnvelope
|
|
handoff = amzn.get("handoff") or self.session_handoff_token
|
|
is_ism = isinstance(getattr(track, "data", None), dict) and "ism" in track.data
|
|
|
|
endpoint = self.endpoints["license_wv"] if widevine else self.endpoints["license_pr"]
|
|
|
|
if self.device_token:
|
|
data_lic = {
|
|
"playbackEnvelope": envelope,
|
|
"licenseChallenge": encoded_challenge,
|
|
"deviceCapabilityFamily": "LivingRoomPlayer",
|
|
"packagingFormat": "SMOOTH_STREAMING" if is_ism else "MPEG_DASH",
|
|
}
|
|
if widevine:
|
|
data_lic["includeHdcpTestKey"] = True
|
|
if getattr(track, "kid", None):
|
|
try:
|
|
data_lic["keyId"] = str(uuid.UUID(str(track.kid))).upper()
|
|
except (ValueError, AttributeError):
|
|
pass
|
|
params = {
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.device.get("device_type", self.config["device_types"]["browser"]),
|
|
"firmware": "1",
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"titleId": title.id,
|
|
"uxLocale": "en_US",
|
|
}
|
|
else:
|
|
if not handoff:
|
|
self.log.error(" - No device token and no sessionHandoffToken; cannot license."); raise SystemExit(1)
|
|
data_lic = {
|
|
"playbackEnvelope": envelope,
|
|
"licenseChallenge": encoded_challenge,
|
|
"deviceCapabilityFamily": "WebPlayer",
|
|
"sessionHandoffToken": handoff,
|
|
}
|
|
if widevine:
|
|
data_lic["includeHdcpTestKey"] = True
|
|
params = {
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.device["device_type"],
|
|
"gascEnabled": str(self.pv).lower(),
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"uxLocale": "en_EN",
|
|
"firmware": 1,
|
|
"titleId": title.id,
|
|
"nerid": self.generate_nerid(),
|
|
}
|
|
|
|
try:
|
|
resp = self.session.post(
|
|
url=endpoint,
|
|
params=params,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self._bearer()}" if self._bearer() else None,
|
|
},
|
|
json=data_lic,
|
|
)
|
|
resp.raise_for_status()
|
|
res = resp.json()
|
|
except requests.exceptions.HTTPError as e:
|
|
msg = "Failed to license"
|
|
status = e.response.status_code if e.response is not None else None
|
|
if e.response is not None:
|
|
try:
|
|
msg += f": {e.response.json()}"
|
|
except Exception:
|
|
msg += f": {e.response.text[:200]}"
|
|
self.log.error(f" - {msg}")
|
|
if status in (403, 500):
|
|
self.log.error(f" - {self._license_denial_hint(track)}")
|
|
raise SystemExit(1)
|
|
except Exception as e:
|
|
self.log.error(f" - Failed to license: {e}"); raise SystemExit(1)
|
|
|
|
if "errorsByResource" in res:
|
|
error = res["errorsByResource"]
|
|
code = error.get("errorCode") or error.get("type") or "Unknown"
|
|
if code == "PRS.NoRights.AnonymizerIP":
|
|
self.log.error(" - Amazon detected a Proxy/VPN and refused to return a license."); raise SystemExit(1)
|
|
self.log.error(f" - Amazon reported an error during the License request: [{code}]")
|
|
self.log.error(f" - {self._license_denial_hint(track)}")
|
|
raise SystemExit(1)
|
|
if "error" in res:
|
|
self.log.error(f" - License Error: {res['error'].get('message', 'Unknown')}"); raise SystemExit(1)
|
|
|
|
primary_key = "widevineLicense" if widevine else "playReadyLicense"
|
|
lic = None
|
|
if isinstance(res.get(primary_key), dict):
|
|
lic = res[primary_key].get("license")
|
|
if not lic:
|
|
for key, val in res.items():
|
|
if "license" in key.lower() and isinstance(val, dict) and val.get("license"):
|
|
lic = val["license"]; break
|
|
if "license" in key.lower() and isinstance(val, str) and val:
|
|
lic = val; break
|
|
if not lic:
|
|
self.log.error(
|
|
f" - License response did not contain a '{primary_key}'. "
|
|
f"Response keys: {list(res.keys())}. Raw (truncated): {json.dumps(res)[:600]}"
|
|
)
|
|
self.log.error(f" - {self._license_denial_hint(track)}")
|
|
raise SystemExit(1)
|
|
return base64.b64decode(lic) if isinstance(lic, str) else lic
|
|
|
|
def _requested_uhd(self) -> bool:
|
|
q = self.quality[0] if isinstance(self.quality, list) and self.quality else self.quality
|
|
return self.vquality == "UHD" or (isinstance(q, int) and q > 1080)
|
|
|
|
def _license_denial_hint(self, track=None) -> str:
|
|
level = None
|
|
try:
|
|
level = getattr(self.cdm, "security_level", None)
|
|
except Exception:
|
|
pass
|
|
cdm_bit = ""
|
|
if level == 3:
|
|
cdm_bit = " Your CDM is Widevine L3 (Amazon licenses only SD to L3)."
|
|
elif level in (2000, 3000):
|
|
cdm_bit = f" Your CDM is PlayReady SL{level}."
|
|
want = []
|
|
if self._requested_uhd() or (isinstance(getattr(track, 'height', None), int) and track.height > 1080):
|
|
want.append("UHD")
|
|
if self.range and self.range != "SDR":
|
|
want.append(self.range)
|
|
want_bit = f" (you requested {'/'.join(want)})" if want else ""
|
|
return (
|
|
f"License failed. CDM robustness limit. {want_bit}.{cdm_bit} "
|
|
"UHD/HDR/DV needs Widevine L1 or PlayReady SL3000. Re-run at a lower quality "
|
|
"(e.g. -q 1080) or SDR to get a licensable stream."
|
|
)
|
|
|
|
def _warn_cdm_quality_mismatch(self) -> None:
|
|
try:
|
|
level = getattr(self.cdm, "security_level", None)
|
|
if level == 3 and (self._requested_uhd() or (self.range and self.range != "SDR")):
|
|
self.log.warning(
|
|
" - Widevine L3 CDM with a UHD/HDR request. Amazon only licenses SD to L3, so the "
|
|
"license will most likely be denied. For UHD/HDR use Widevine L1 / PlayReady SL3000, or -q 1080 / SDR."
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
def _tag_amzn_tracks(self, track_iterable, manifest: dict) -> None:
|
|
token = (manifest.get("sessionization") or {}).get("sessionHandoffToken")
|
|
envelope = self.playbackEnvelope
|
|
for t in track_iterable:
|
|
try:
|
|
if not isinstance(t.data, dict):
|
|
continue
|
|
amzn = t.data.setdefault("amzn", {})
|
|
if token:
|
|
amzn["handoff"] = token
|
|
if envelope:
|
|
amzn["envelope"] = envelope
|
|
except Exception:
|
|
pass
|
|
|
|
def _post_process_audio(self, audio_tracks) -> None:
|
|
for audio in audio_tracks:
|
|
try:
|
|
data = getattr(audio, "data", None)
|
|
if not isinstance(data, dict) or "dash" not in data:
|
|
continue
|
|
adaptation_set = data["dash"].get("adaptation_set")
|
|
representation = data["dash"].get("representation")
|
|
if adaptation_set is None:
|
|
continue
|
|
|
|
subtype = (adaptation_set.get("audioTrackSubtype") or "").lower()
|
|
if "descriptive" in subtype:
|
|
audio.descriptive = True
|
|
if "boosteddialog" in subtype:
|
|
audio.bitrate = 1
|
|
if not audio.joc:
|
|
for elem in (adaptation_set, representation):
|
|
if elem is None:
|
|
continue
|
|
try:
|
|
props = elem.findall("SupplementalProperty") + elem.findall("EssentialProperty")
|
|
except Exception:
|
|
props = []
|
|
for prop in props:
|
|
scheme = prop.get("schemeIdUri", "")
|
|
value = (prop.get("value", "") or "").upper()
|
|
if scheme == "tag:dolby.com,2018:dash:EC3_ExtensionType:2018" and value == "JOC":
|
|
audio.joc = 16
|
|
break
|
|
if audio.joc:
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
def manage_session(self, track) -> None:
|
|
try:
|
|
handoff = None
|
|
if isinstance(getattr(track, "data", None), dict):
|
|
handoff = (track.data.get("amzn") or {}).get("handoff")
|
|
handoff = handoff or self.session_handoff_token
|
|
if not handoff or not isinstance(handoff, str):
|
|
self.log.debug("No session handoff token available; skipping keep-alive session")
|
|
return
|
|
|
|
refreshed = self.playbackEnvelope_update(self.playbackInfo) or self.playbackInfo or {}
|
|
envelope = (refreshed.get("playbackExperienceMetadata", {}) or {}).get("playbackEnvelope") \
|
|
or self.playbackEnvelope
|
|
if not envelope:
|
|
self.log.debug("No playback envelope for keep-alive session; skipping")
|
|
return
|
|
|
|
def session_params():
|
|
return {
|
|
"deviceID": self.device_id,
|
|
"deviceTypeID": self.device["device_type"],
|
|
"gascEnabled": str(self.pv).lower(),
|
|
"marketplaceID": self.region["marketplace_id"],
|
|
"uxLocale": "en_EN",
|
|
"firmware": "1",
|
|
"version": "1",
|
|
"nerid": self.generate_nerid(),
|
|
}
|
|
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"accept": "application/json",
|
|
"x-request-priority": "CRITICAL",
|
|
"x-retry-count": "0",
|
|
}
|
|
|
|
progress = round(random.uniform(0, 10), 6)
|
|
step = 3
|
|
start_time = datetime.utcnow().isoformat(timespec="milliseconds") + "Z"
|
|
|
|
res = self.session.post(
|
|
url=self.endpoints["opensession"],
|
|
params=session_params(),
|
|
headers=headers,
|
|
json={
|
|
"sessionHandoff": handoff,
|
|
"playbackEnvelope": envelope,
|
|
"streamInfo": {
|
|
"eventType": "START",
|
|
"streamUpdateTime": progress,
|
|
"vodProgressInfo": {
|
|
"currentProgressTime": f"PT{progress:.6f}S",
|
|
"timeFormat": "ISO8601DURATION",
|
|
},
|
|
},
|
|
"userWatchSessionId": str(uuid.uuid4()),
|
|
},
|
|
)
|
|
if res.status_code != 200:
|
|
self.log.debug(f"Keep-alive: unable to open session ({res.status_code})")
|
|
return
|
|
session_token = res.json().get("sessionToken")
|
|
if not session_token:
|
|
return
|
|
|
|
time.sleep(step)
|
|
update_time = (datetime.fromisoformat(start_time[:-1]) + timedelta(seconds=step)) \
|
|
.isoformat(timespec="milliseconds") + "Z"
|
|
|
|
res = self.session.post(
|
|
url=self.endpoints["updatesession"],
|
|
params=session_params(),
|
|
headers=headers,
|
|
json={
|
|
"sessionToken": session_token,
|
|
"streamInfo": {
|
|
"eventType": "PAUSE",
|
|
"streamUpdateTime": update_time,
|
|
"vodProgressInfo": {
|
|
"currentProgressTime": f"PT{progress + step:.6f}S",
|
|
"timeFormat": "ISO8601DURATION",
|
|
},
|
|
},
|
|
},
|
|
)
|
|
if res.status_code != 200:
|
|
self.log.debug(f"Keep-alive: unable to update session ({res.status_code})")
|
|
return
|
|
session_token = res.json().get("sessionToken", session_token)
|
|
|
|
res = self.session.post(
|
|
url=self.endpoints["closesession"],
|
|
params=session_params(),
|
|
headers=headers,
|
|
json={
|
|
"sessionToken": session_token,
|
|
"streamInfo": {
|
|
"eventType": "STOP",
|
|
"streamUpdateTime": update_time,
|
|
"vodProgressInfo": {
|
|
"currentProgressTime": f"PT{progress + step:.6f}S",
|
|
"timeFormat": "ISO8601DURATION",
|
|
},
|
|
},
|
|
},
|
|
)
|
|
if res.status_code == 200:
|
|
self.log.info(" + True-region playback session completed")
|
|
else:
|
|
self.log.debug(f"Keep-alive: unable to close session ({res.status_code})")
|
|
except Exception as e:
|
|
self.log.debug(f"Keep-alive session failed: {e}")
|
|
|
|
@staticmethod
|
|
def _usable_manifest(manifest: dict) -> bool:
|
|
return bool(
|
|
(manifest or {}).get("vodPlaybackUrls", {}).get("result", {})
|
|
.get("playbackUrls", {}).get("urlSets")
|
|
)
|
|
|
|
def choose_manifest(self, manifest: dict, cdn=None):
|
|
if not manifest or "vodPlaybackUrls" not in manifest:
|
|
return {}
|
|
|
|
url_sets = (
|
|
manifest["vodPlaybackUrls"].get("result", {}).get("playbackUrls", {}).get("urlSets", [])
|
|
)
|
|
if not url_sets:
|
|
return {}
|
|
|
|
if cdn:
|
|
cdn = cdn.lower()
|
|
return next((x for x in url_sets if (x.get("cdn") or "").lower() == cdn), {})
|
|
akamai = [x for x in url_sets if "akamai" in (x.get("cdn") or "").lower()]
|
|
if akamai:
|
|
return secrets.choice(akamai)
|
|
return secrets.choice(url_sets)
|
|
|
|
@staticmethod
|
|
def generate_nerid(length=24):
|
|
chars = string.ascii_letters + string.digits
|
|
return "".join(secrets.choice(chars) for _ in range(length))
|
|
|
|
@staticmethod
|
|
def clean_mpd_url(mpd_url, optimise=False):
|
|
if optimise:
|
|
return mpd_url.replace("~", "") + "?encoding=segmentBase"
|
|
if match := re.match(r"(https?://.*/)d.?/.*~/(.*)", mpd_url):
|
|
mpd_url = "".join(match.groups())
|
|
else:
|
|
try:
|
|
mpd_url = "".join(
|
|
re.split(r"(?i)(/)", mpd_url)[:5] + re.split(r"(?i)(/)", mpd_url)[9:]
|
|
)
|
|
except IndexError:
|
|
pass
|
|
return mpd_url
|
|
|
|
def get_region(self) -> dict:
|
|
domain_region = self.get_domain_region()
|
|
if not domain_region:
|
|
return {}
|
|
|
|
region = self.config["regions"].get(domain_region)
|
|
if not region:
|
|
self.log.error(f" - There's no region configuration data for the region: {domain_region}"); raise SystemExit(1)
|
|
|
|
region["code"] = domain_region
|
|
|
|
if self.pv:
|
|
res = self.session.get("https://www.primevideo.com").text
|
|
match = re.search(r'ue_furl *= *([\'"])fls-(na|eu|fe)\.amazon\.[a-z.]+\1', res)
|
|
if match:
|
|
pv_region = match.group(2).lower()
|
|
else:
|
|
self.log.error(" - Failed to get PrimeVideo region"); raise SystemExit(1)
|
|
pv_region = {"na": "atv-ps"}.get(pv_region, f"atv-ps-{pv_region}")
|
|
region["base_manifest"] = f"{pv_region}.primevideo.com"
|
|
region["base"] = "www.primevideo.com"
|
|
|
|
return region
|
|
|
|
def get_domain_region(self):
|
|
tlds = [tldextract.extract(x.domain) for x in self.session.cookies if x.domain_specified]
|
|
tld = next((x.suffix for x in tlds if x.domain.lower() in ("amazon", "primevideo")), None)
|
|
if tld:
|
|
tld = tld.split(".")[-1]
|
|
region = {"com": "us", "uk": "gb"}.get(tld, tld)
|
|
|
|
if region == "us":
|
|
lc_cookie = next(
|
|
(x.value for x in self.session.cookies
|
|
if x.name in ("lc-main-av", "lc-main") and x.domain_specified),
|
|
None
|
|
)
|
|
if lc_cookie:
|
|
parts = lc_cookie.replace("-", "_").split("_")
|
|
if len(parts) >= 2:
|
|
country = parts[-1].lower()
|
|
if country not in ("us", ""):
|
|
mapped = {"uk": "gb"}.get(country, country)
|
|
if mapped in self.config.get("regions", {}):
|
|
region = mapped
|
|
|
|
return region
|
|
|
|
def prepare_endpoint(self, name: str, uri: str, region: dict) -> str:
|
|
if name in ("playback", "license_wv", "license_pr", "xray",
|
|
"refreshplayback", "opensession", "updatesession", "closesession",
|
|
"configuration"):
|
|
return f"https://{region['base_manifest']}{uri}"
|
|
if name in ("ontv", "devicelink", "details", "getDetailWidgets", "metadata"):
|
|
if self.pv:
|
|
host = "www.primevideo.com"
|
|
else:
|
|
host = f"{region['base']}/gp/video" if name == "metadata" else region["base"]
|
|
return f"https://{host}{uri}"
|
|
if name in ("codepair", "register", "token"):
|
|
base_api = region.get("base_api") or self.config["regions"]["us"]["base_api"]
|
|
return f"https://{base_api}{uri}"
|
|
raise ValueError(f"Unknown endpoint: {name}")
|
|
|
|
def prepare_endpoints(self, endpoints: dict, region: dict) -> dict:
|
|
return {k: self.prepare_endpoint(k, v, region) for k, v in endpoints.items()}
|
|
|
|
def register_device(self) -> None:
|
|
_profile = self.profile or "default"
|
|
self.device = dict((self.config.get("device") or {}).get(_profile, {}))
|
|
|
|
identity_cache = Cacher("AMZN")
|
|
identity_key = f"device_identity_{_profile}"
|
|
cached_identity = identity_cache.get(identity_key)
|
|
|
|
if cached_identity and cached_identity.data:
|
|
identity = cached_identity.data
|
|
self.log.debug(" + Using cached device identity")
|
|
else:
|
|
unique_serial = secrets.token_hex(8)
|
|
base_name = self.device.get("device_name", "%FIRST_NAME%'s Shield TV")
|
|
clean_name = re.sub(r"%DUPE_STRATEGY[^%]*%", "", base_name).rstrip()
|
|
suffix = secrets.token_hex(2).upper()
|
|
unique_name = f"{clean_name}-{suffix}"
|
|
identity = {"device_serial": unique_serial, "device_name": unique_name}
|
|
cached_identity = identity_cache.get(identity_key)
|
|
cached_identity.set(identity, int(time.time()) + 60 * 60 * 24 * 3650)
|
|
self.log.info(f" + Generated unique device identity: serial={unique_serial}, name={unique_name!r}")
|
|
|
|
self.device["device_serial"] = identity["device_serial"]
|
|
self.device["device_name"] = identity["device_name"]
|
|
|
|
device_hash = hashlib.md5(json.dumps(self.device, sort_keys=True).encode()).hexdigest()[0:6]
|
|
device_cache_path = f"device_tokens_{_profile}_{device_hash}"
|
|
|
|
_reg = self.DeviceRegistration(
|
|
device=self.device,
|
|
endpoints=self.endpoints,
|
|
log=self.log,
|
|
cache_path=device_cache_path,
|
|
session=self.session
|
|
)
|
|
self.device_token = _reg.bearer
|
|
self.device_refresh_token = _reg.refresh_token
|
|
|
|
self.device_id = self.device.get("device_serial")
|
|
if not self.device_id:
|
|
self.log.error(f" - A device serial is required in the config, try: {os.urandom(8).hex()}"); raise SystemExit(1)
|
|
|
|
class DeviceRegistration:
|
|
def __init__(self, device: dict, endpoints: dict, cache_path: str, session: requests.Session, log):
|
|
self.session = session
|
|
self.device = device
|
|
self.endpoints = endpoints
|
|
self.cache_path = cache_path
|
|
self.log = log
|
|
self.cache = Cacher("AMZN")
|
|
|
|
self.device = {k: str(v) if not isinstance(v, str) else v for k, v in self.device.items()}
|
|
self.bearer = None
|
|
self.refresh_token = None
|
|
|
|
cached_data = self.cache.get(self.cache_path)
|
|
|
|
if cached_data and cached_data.data:
|
|
if cached_data.data.get("expires_in", 0) > int(time.time()):
|
|
self.log.info(" + Using cached device bearer")
|
|
self.bearer = cached_data.data["access_token"]
|
|
self.refresh_token = cached_data.data.get("refresh_token")
|
|
else:
|
|
self.log.info("Cached device bearer expired, refreshing...")
|
|
refresh_token = cached_data.data.get("refresh_token")
|
|
refreshed_tokens = self.refresh(self.device, refresh_token) if refresh_token else None
|
|
if refreshed_tokens and refreshed_tokens.get("access_token"):
|
|
refreshed_tokens["refresh_token"] = refreshed_tokens.get("refresh_token") or refresh_token
|
|
expires_seconds = int(refreshed_tokens.get("expires_in") or 3600)
|
|
refreshed_tokens["expires_in"] = int(time.time()) + expires_seconds
|
|
cached_data.set(refreshed_tokens, refreshed_tokens["expires_in"])
|
|
self.bearer = refreshed_tokens["access_token"]
|
|
self.refresh_token = refreshed_tokens.get("refresh_token")
|
|
else:
|
|
self.log.info(" + Re-registering device bearer")
|
|
self.bearer = self.register(self.device)
|
|
else:
|
|
self.log.info(" + Registering new device bearer")
|
|
self.bearer = self.register(self.device)
|
|
|
|
def register(self, device: dict) -> str:
|
|
code_pair = self.get_code_pair(device)
|
|
public_code = code_pair["public_code"]
|
|
|
|
self.log.info(f" + Visit https://www.primevideo.com/mytv and enter code: {public_code}")
|
|
self.log.info(" Waiting for authorisation (May take up to 5 minutes)")
|
|
|
|
interval = 10
|
|
deadline = int(time.time()) + 300
|
|
|
|
while int(time.time()) < deadline:
|
|
res = self.session.post(
|
|
url=self.endpoints["register"],
|
|
headers={"Content-Type": "application/json", "Accept-Language": "en-US"},
|
|
json={
|
|
"auth_data": {"code_pair": code_pair},
|
|
"registration_data": device,
|
|
"requested_token_type": ["bearer"],
|
|
"requested_extensions": ["device_info", "customer_info"]
|
|
},
|
|
cookies=None
|
|
)
|
|
data = res.json()
|
|
|
|
if res.status_code == 200 and "success" in data.get("response", {}):
|
|
break
|
|
|
|
error_code = data.get("response", {}).get("error", {}).get("code", "")
|
|
if error_code == "Unauthorized":
|
|
time.sleep(interval)
|
|
continue
|
|
else:
|
|
self.log.error(f"Unable to register: {res.text}"); raise SystemExit(1)
|
|
else:
|
|
self.log.error("Device registration timed out. Code not approved in time."); raise SystemExit(1)
|
|
|
|
bearer = data["response"]["success"]["tokens"]["bearer"]
|
|
expires_val = bearer.get("expires_in", 3600)
|
|
if isinstance(expires_val, dict):
|
|
expires_val = expires_val.get("value", 3600)
|
|
bearer_data = {
|
|
"access_token": bearer["access_token"],
|
|
"refresh_token": bearer.get("refresh_token", ""),
|
|
"expires_in": int(time.time()) + int(expires_val),
|
|
}
|
|
keyed_cache = self.cache.get(self.cache_path)
|
|
keyed_cache.set(bearer_data, int(time.time()) + int(expires_val))
|
|
|
|
self.refresh_token = bearer_data["refresh_token"]
|
|
self.log.info(" + Device registered and token cached successfully")
|
|
return bearer_data["access_token"]
|
|
|
|
def refresh(self, device: dict, refresh_token: str):
|
|
try:
|
|
res = self.session.post(
|
|
url=self.endpoints["token"],
|
|
json={
|
|
"app_name": device["app_name"],
|
|
"app_version": device["app_version"],
|
|
"source_token_type": "refresh_token",
|
|
"source_token": refresh_token,
|
|
"requested_token_type": "access_token"
|
|
}
|
|
).json()
|
|
except Exception as e:
|
|
self.log.warning(f"Device token refresh request failed ({e}); re-registering.")
|
|
return None
|
|
|
|
if "error" in res:
|
|
self.log.warning(
|
|
f"Could not refresh device token ({res.get('error_description') or res.get('error')}); re-registering."
|
|
)
|
|
return None
|
|
|
|
return res
|
|
|
|
def get_csrf_token(self) -> str:
|
|
res = self.session.get(self.endpoints["ontv"])
|
|
if 'name="appAction" value="SIGNIN"' in res.text or "SIGNIN_PWD_COLLECT" in res.text:
|
|
self.log.error("Cookies are signed out, cannot get ontv CSRF token.")
|
|
raise SystemExit(1)
|
|
for match in re.finditer(r'<script type="text/template">(.+?)</script>', res.text, re.DOTALL):
|
|
try:
|
|
prop = json.loads(match.group(1))
|
|
token = prop.get("props", {}).get("codeEntry", {}).get("token")
|
|
if token:
|
|
return token
|
|
token = prop.get("codeEntry", {}).get("token")
|
|
if token:
|
|
return token
|
|
except Exception:
|
|
pass
|
|
ce_idx = res.text.find('"codeEntry"')
|
|
if ce_idx != -1:
|
|
snippet = res.text[ce_idx:ce_idx + 2000]
|
|
m2 = re.search(r'"token"\s*:\s*"([^"]+)"', snippet)
|
|
if m2:
|
|
return m2.group(1)
|
|
self.log.error("Unable to get ontv CSRF token")
|
|
raise SystemExit(1)
|
|
|
|
def get_code_pair(self, device: dict) -> dict:
|
|
res = self.session.post(
|
|
url=self.endpoints["codepair"],
|
|
headers={"Content-Type": "application/json", "Accept-Language": "en-US"},
|
|
json={"code_data": device}
|
|
).json()
|
|
if "error" in res:
|
|
self.log.error(f"Unable to get code pair: {res['error']}"); raise SystemExit(1)
|
|
return res
|
|
|
|
def parse_title(self, ctx, title):
|
|
title = title or ctx.parent.params.get("title")
|
|
if not title:
|
|
self.log.error(" - No title ID specified")
|
|
if not getattr(self, "TITLE_RE"):
|
|
self.title = title
|
|
return {}
|
|
for regex in as_list(self.TITLE_RE):
|
|
m = re.search(regex, title)
|
|
if m:
|
|
self.title = m.group("id")
|
|
return m.groupdict()
|
|
self.log.warning(f" - Unable to parse title ID {title!r}, using as-is")
|
|
self.title = title
|