Untitled
unknown
plain_text
22 days ago
8.2 kB
3
Indexable
Never
from __future__ import annotations import base64 import json import os import re from datetime import datetime from typing import Any, Optional, Union from urllib.parse import unquote import click import m3u8 from click import Context from pymp4.parser import Box from vinetrimmer.objects import AudioTrack, MenuTrack, TextTrack, Title, Track, Tracks from vinetrimmer.services.BaseService import BaseService from vinetrimmer.utils.collections import as_list class AppleTVPlus(BaseService): """ Service code for Apple's TV Plus streaming service (https://tv.apple.com). \b WIP: decrypt and removal of bumper/dub cards \b Authorization: Cookies Security: UHD@L1 FHD@L1 HD@L3 """ ALIASES = ["ATVP", "appletvplus", "appletv+"] #GEOFENCE = ["us"] VIDEO_CODEC_MAP = { "H264": ["avc"], "H265": ["hvc", "hev", "dvh"] } AUDIO_CODEC_MAP = { "AAC": ["HE", "stereo"], "AC3": ["ac3"], "EC3": ["ec3", "atmos"] } @staticmethod @click.command(name="AppleTVPlus", short_help="https://tv.apple.com") @click.argument("title", type=str) @click.pass_context def cli(ctx: Context, **kwargs: Any) -> AppleTVPlus: return AppleTVPlus(ctx, **kwargs) def __init__(self, ctx: Context, title: str): self.title = self.get_url_title(title) super().__init__(ctx) assert ctx.parent is not None self.vcodec = ctx.parent.params["vcodec"] self.acodec = ctx.parent.params["acodec"] self.extra_server_parameters = None self.configure() def get_url_title(self, title: str) -> str: """ Get title from url """ if 'https:' in title: return os.path.basename(title.split('?')[0]) else: return title def get_titles(self) -> Union[Title, list[Title]]: r = None for i in range(2): r = self.session.get( url=self.config["endpoints"]["title"].format(type={0: "shows", 1: "movies"}[i], id=self.title), params=self.config["device"] ) if r.status_code != 404: break if not r: self.log.exit(f" - Title ID '{self.title}' could not be found.") raise try: title_information = r.json()["data"]["content"] except json.JSONDecodeError: raise ValueError(f"Failed to load title manifest: {r.text}") if title_information["type"] == "Movie": return Title( id_=self.title, type_=Title.Types.MOVIE, name=title_information["title"], year=datetime.utcfromtimestamp(title_information["releaseDate"] / 1000).year, original_lang=title_information["originalSpokenLanguages"][0]["locale"], source=self.ALIASES[0], service_data=title_information ) r = self.session.get( url=self.config["endpoints"]["tv_episodes"].format(id=self.title), params=self.config["device"] ) try: episodes = r.json()["data"]["episodes"] except json.JSONDecodeError: raise ValueError(f"Failed to load episodes list: {r.text}") return [Title( id_=self.title, type_=Title.Types.TV, name=episode["showTitle"], season=episode["seasonNumber"], episode=episode["episodeNumber"], episode_name=episode.get("title"), original_lang=title_information["originalSpokenLanguages"][0]["locale"], source=self.ALIASES[0], service_data=episode ) for episode in episodes] def get_tracks(self, title: Title) -> Tracks: res = self.session.get( url=self.config["endpoints"]["manifest"].format(id=title.service_data["id"]), params=self.config["device"] ) try: stream_data = res.json() except json.JSONDecodeError: raise ValueError(f"Failed to load stream data: {res.text}") stream_data = stream_data["data"]["content"]["playables"][0] if not stream_data["isEntitledToPlay"]: self.log.exit(" - User is not entitled to play this title") raise headers = list(self.session.headers.items()) self.session.headers.clear() self.extra_server_parameters = stream_data["assets"]["fpsKeyServerQueryParameters"] r = self.session.get(stream_data["assets"]["hlsUrl"], headers={'User-Agent': 'AppleTV11,1/11.1'}) self.session.headers.update(headers) res = r.text #print(r.url) tracks = Tracks.from_m3u8( master=m3u8.loads(res, r.url), lang=title.original_lang, source=self.ALIASES[0], accesstkn=self.get_environment_config()['MEDIA_API']['token'], extraSeverParamters=self.extra_server_parameters, mediaUserToken=self.session.cookies.get_dict()["media-user-token"] ) for track in tracks: track_data = track.extra # type: Union[vinetrimmer.utils.m3u8forApple.Media, vinetrimmer.utils.m3u8forApple.Playlist] if isinstance(track, AudioTrack): track.encrypted = True bitrate = re.search(r"&g=(\d+?)&", track_data.uri) if bitrate: track.bitrate = int(bitrate[1][-3::]) * 1000 # e.g. 128->128,000, 2448->448,000 else: raise ValueError(f"Unable to get a bitrate value for Track {track.id}") track.codec = track.codec.replace("_vod", "") if isinstance(track, TextTrack): track.codec = "vtt" tracks.videos = [ x for x in tracks.videos if x.codec[:3] in self.VIDEO_CODEC_MAP[self.vcodec] ] if self.acodec: tracks.audio = [ x for x in tracks.audio if x.codec.split("-")[0] in self.AUDIO_CODEC_MAP[self.acodec] ] sdh_tracks = [x.language for x in tracks.subtitles if x.sdh] tracks.subtitles = [x for x in tracks.subtitles if x.language not in sdh_tracks or x.sdh or x.forced] return Tracks([ # multiple CDNs, only want one x for x in tracks if any( cdn in as_list(x.url)[0].split("?")[1].split("&") for cdn in ["cdn=ak", "cdn=vod-ak-aoc.tv.apple.com"] ) ]) def get_chapters(self, title: Title) -> list[MenuTrack]: return [] def certificate(self, **_: Any) -> None: return None # will use common privacy cert def license(self, challenge: bytes, track: Track, **_: Any) -> Any: xxxx XXXX ).json() except requests.HTTPError as e: print(e) if not e.response.text: raise self.log.exit(" - No license returned!") raise self.log.exit(f" - Unable to obtain license (error code: {e.response.json()['errorCode']})") return res['streaming-response']['streaming-keys'][0]["license"] # Service specific functions def configure(self) -> None: environment = self.get_environment_config() if not environment: raise ValueError("Failed to get AppleTV+ WEB TV App Environment Configuration...") self.session.headers.update({ "User-Agent": self.config["user_agent"], "Authorization": f"Bearer {environment['MEDIA_API']['token']}", "media-user-token": self.session.cookies.get_dict()["media-user-token"], "x-apple-music-user-token": self.session.cookies.get_dict()["media-user-token"] }) def get_environment_config(self) -> Optional[dict]: """Loads environment config data from WEB App's <meta> tag.""" res = self.session.get("https://tv.apple.com").text env = re.search(r'web-tv-app/config/environment"[\s\S]*?content="([^"]+)', res) if not env: return None return json.loads(unquote(env[1]))
Leave a Comment