thegame_1980/TwinVine
1
0
Fork 0
mirror of https://github.com/vinefeeder/TwinVine.git synced 2026-07-15 18:10:04 +02:00
Code Issues Packages Projects Releases Wiki Activity

service updates

Browse Source
This commit is contained in:
VineFeeder
2025-11-20 09:36:45 +00:00
parent e75ab373a5
commit a479ac5b1b
9 changed files with 1661 additions and 251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/envied/src/envied/services/DSCP/__init__.py
+425 -143
View File
@@ -2,28 +2,34 @@ from __future__ import annotations
import json
import re
import sys
import uuid
from collections import defaultdict
from collections.abc import Generator
from copy import deepcopy
from http.cookiejar import CookieJar
from typing import Any, Optional, Union
from typing import Any
from urllib.parse import urljoin
from zlib import crc32
import click
from click import Context
from langcodes import Language
from lxml import etree
from envied.core.credential import Credential
from envied.core.manifests import DASH, HLS
from envied.core.manifests import DASH
from envied.core.search_result import SearchResult
from envied.core.service import Service
from envied.core.session import session as CurlSession
from envied.core.titles import Episode, Movie, Movies, Series
from envied.core.tracks import Chapters, Tracks
from requests import Request
from envied.core.tracks import Audio, Chapter, Chapters, Subtitle, Track, Tracks
from envied.core.utilities import is_close_match
class DSCP(Service):
"""
\b
Service code for Discovery Plus (https://discoveryplus.com).
Service code for Discovery Plus streaming service (https://www.discoveryplus.com).
Credit to @sp4rk.y for the subtitle fix.
\b
Version: 1.0.0
@@ -31,60 +37,106 @@ class DSCP(Service):
Authorization: Cookies
Robustness:
Widevine:
L3: 2160p, AAC2.0
ClearKey:
AES-128: 1080p, AAC2.0
L1: 2160p, 1080p
L3: 720p
PlayReady:
SL3000: 2160p
SL2000: 1080p, 720p
\b
Tips:
- Input can be either complete title URL or just the path:
SHOW: /show/richard-hammonds-workshop
EPISODE: /video/richard-hammonds-workshop/new-beginnings
SPORT: /video/sport/tnt-sports-1/uefa-champions-league
- Use the --lang LANG_RANGE option to request non-english tracks
- use -v H.265 to request H.265 UHD tracks (if available)
- Input examples:
SHOW: https://play.discoveryplus.com/show/eb26e00e-9582-4790-a61c-48d785926f58
STANDALONE: https://play.discoveryplus.com/standalone/5012ae3f-d9bd-46ec-ad42-b8116b811441
SPORT: https://play.discoveryplus.com/sport/9cc449de-2a64-524d-bcb6-cabd4ac70340
EPISODE: https://play.discoveryplus.com/video/watch/8685efdd-a3c4-4892-b1d1-5f9f071cacf1/de67ea8e-a90f-4609-81af-4f09906f60b2
\b
Notes:
- Using '-v H.265' will request DASH manifest even if no H.265 tracks are available.
This can be useful if HLS is not available for some reason.
- Language tags can be mislabelled or missing on some titles. List tracks with --list to verify.
- All qualities, codecs, and ranges are included when available. Use -v H.265, -r HDR10, -q 1080p, etc. to select.
\b
Bonus tip: With some minor adjustments to the code and config, you can convert this to an HMAX service.
- Replace all instances of "DSCP" with "HMAX"
- Replace all instances of "dplus" with "beam"
- Replace all instances of "discoveryplus" with "hbomax"
"""
ALIASES = ("dplus", "discoveryplus", "discovery+")
TITLE_RE = r"^(?:https?://(?:www\.)?discoveryplus\.com(?:/[a-z]{2})?)?/(?P<type>show|video)/(?P<id>[a-z0-9-/]+)"
ALIASES = ("discoveryplus",)
TITLE_RE = (
r"^(?:https?://play.discoveryplus\.com?)?/(?P<type>show|mini-series|video|movie|topical|standalone|sport)/(?P<id>[a-z0-9-/]+)"
)
@staticmethod
@click.command(name="DSCP", short_help="https://discoveryplus.com", help=__doc__)
@click.command(name="DSCP", short_help="https://www.discoveryplus.com/", help=__doc__)
@click.argument("title", type=str)
@click.pass_context
def cli(ctx: Context, **kwargs: Any) -> DSCP:
return DSCP(ctx, **kwargs)
def __init__(self, ctx: Context, title: str):
self.title = title
self.vcodec = ctx.parent.params.get("vcodec")
super().__init__(ctx)
self.title = title
def authenticate(
self,
cookies: Optional[CookieJar] = None,
credential: Optional[Credential] = None,
) -> None:
self.profile = ctx.parent.params.get("profile")
if not self.profile:
self.profile = "default"
self.cdm = ctx.obj.cdm
if self.cdm is not None:
self.drm_system = "playready"
self.security_level = "SL3000"
if self.cdm.security_level <= 3:
self.drm_system = "widevine"
self.security_level = "L1"
def get_session(self) -> CurlSession:
return CurlSession("okhttp4", status_forcelist=[429, 502, 503, 504])
def authenticate(self, cookies: CookieJar | None = None, credential: Credential | None = None) -> None:
super().authenticate(cookies, credential)
if not cookies:
raise EnvironmentError("Service requires Cookies for Authentication.")
self.session.cookies.update(cookies)
cache = self.cache.get(f"tokens_{self.profile}")
if cache:
self.log.info(" + Using cached Tokens...")
tokens = cache.data
else:
self.log.info(" + Setting up new profile...")
self.base_url = None
info = self._request("GET", "https://global-prod.disco-api.com/bootstrapInfo")
self.base_url = info["data"]["attributes"].get("baseApiUrl")
st_token = self.session.cookies.get_dict().get("st")
if not st_token:
raise ValueError("- Unable to find token in cookies, try refreshing.")
profile = {"token": st_token, "device_id": str(uuid.uuid1())}
cache.set(profile)
tokens = cache.data
self.device_id = tokens["device_id"]
client_id = self.config["client_id"]
self.session.headers.update({
"user-agent": "androidtv dplus/20.8.1.2 (android/9; en-US; SHIELD Android TV-NVIDIA; Build/1)",
"x-disco-client": "ANDROIDTV:9:dplus:20.8.1.2",
"x-disco-params": "realm=bolt,bid=dplus,features=ar",
"x-device-info": f"dplus/20.8.1.2 (NVIDIA/SHIELD Android TV; android/9-mdarcy; {self.device_id}/{client_id})",
})
self.base_url = self.config["endpoints"]["base_url"].format("any", "any")
access = self._request("GET", "/token", params={"realm": "bolt", "deviceId": self.device_id})
self.access_token = access["data"]["attributes"]["token"]
config = self._request("POST", "/session-context/headwaiter/v1/bootstrap")
self.base_url = self.config["endpoints"]["base_url"].format(config["routing"]["tenant"], config["routing"]["homeMarket"])
user = self._request("GET", "/users/me")
self.territory = user["data"]["attributes"]["currentLocationTerritory"]
self.user_language = user["data"]["attributes"]["clientTranslationLanguageTags"][0]
self.site_id = user["meta"]["site"]["id"]
if user["data"]["attributes"]["anonymous"]:
raise ValueError("Anonymous user - try refreshing cookies.")
def search(self) -> Generator[SearchResult, None, None]:
params = {
@@ -100,108 +152,318 @@ class DSCP(Service):
for result in results:
yield SearchResult(
id_=f"/show/{result.get('alternateId')}",
id_=f"https://play.discoveryplus.com/show/{result.get('alternateId')}",
title=result.get("name"),
description=result.get("description"),
label="show",
url=f"/show/{result.get('alternateId')}",
url=f"https://play.discoveryplus.com/show/{result.get('alternateId')}",
)
def get_titles(self) -> Union[Movies, Series]:
def get_titles(self) -> Movies | Series:
try:
kind, content_id = (re.match(self.TITLE_RE, self.title).group(i) for i in ("type", "id"))
entity, content_id = (re.match(self.TITLE_RE, self.title).group(i) for i in ("type", "id"))
except Exception:
raise ValueError("Could not parse ID from title - is the URL correct?")
if kind == "video":
episodes = self._episode(content_id)
if kind == "show":
if entity in ("show", "mini-series", "topical"):
episodes = self._show(content_id)
return Series(episodes)
return Series(episodes)
elif entity in ("movie", "standalone"):
movie = self._movie(content_id, entity)
return Movies(movie)
elif entity == "sport":
sport = self._sport(content_id)
return Movies(sport)
def get_tracks(self, title: Union[Movie, Episode]) -> Tracks:
elif entity == "video":
episodes = self._episode(content_id)
return Series(episodes)
else:
raise ValueError(f"Unknown content: {entity}")
def get_tracks(self, title: Movie | Episode) -> Tracks:
payload = {
"videoId": title.id,
"deviceInfo": {
"adBlocker": "false",
"drmSupported": "false",
"hwDecodingCapabilities": ["H264", "H265"],
"screen": {"width": 3840, "height": 2160},
"player": {"width": 3840, "height": 2160},
"appBundle": "com.wbd.stream",
"applicationSessionId": self.device_id,
"capabilities": {
"codecs": {
"audio": {
"decoders": [
{"codec": "aac", "profiles": ["lc", "he", "hev2", "xhe"]},
{"codec": "eac3", "profiles": ["atmos"]},
]
},
"video": {
"decoders": [
{
"codec": "h264",
"levelConstraints": {
"framerate": {"max": 60, "min": 0},
"height": {"max": 2160, "min": 48},
"width": {"max": 3840, "min": 48},
},
"maxLevel": "5.2",
"profiles": ["baseline", "main", "high"],
},
{
"codec": "h265",
"levelConstraints": {
"framerate": {"max": 60, "min": 0},
"height": {"max": 2160, "min": 144},
"width": {"max": 3840, "min": 144},
},
"maxLevel": "5.1",
"profiles": ["main10", "main"],
},
],
"hdrFormats": ["hdr10", "hdr10plus", "dolbyvision", "dolbyvision5", "dolbyvision8", "hlg"],
},
},
"contentProtection": {
"contentDecryptionModules": [
{"drmKeySystem": self.drm_system, "maxSecurityLevel": self.security_level}
]
},
"manifests": {"formats": {"dash": {}}},
},
"wisteriaProperties": {
"product": "dplus_emea",
"sessionId": str(uuid.uuid1()),
"consumptionType": "streaming",
"deviceInfo": {
"player": {
"mediaEngine": {"name": "", "version": ""},
"playerView": {"height": 2160, "width": 3840},
"sdk": {"name": "", "version": ""},
}
},
"editId": title.id,
"firstPlay": False,
"gdpr": False,
"playbackSessionId": str(uuid.uuid4()),
"userPreferences": {
#'uiLanguage': 'en'
},
}
if self.vcodec == "H.265":
payload["wisteriaProperties"]["device"] = {
"browser": {"name": "chrome", "version": "96.0.4664.55"},
"type": "firetv",
}
payload["wisteriaProperties"]["platform"] = "firetv"
playback = self._request(
"POST", "/playback-orchestrator/any/playback-orchestrator/v1/playbackInfo",
headers={"Authorization": f"Bearer {self.access_token}"},
json=payload,
)
original_language = next((
x.get("language")
for x in playback["videos"][0]["audioTracks"]
if "Original" in x.get("displayName", "")
), "")
res = self._request("POST", "/playback/v3/videoPlaybackInfo", payload=payload)
manifest = (
playback.get("fallback", {}).get("manifest", {}).get("url", "").replace("_fallback", "")
or playback.get("manifest", {}).get("url")
)
streaming = res["data"]["attributes"]["streaming"][0]
streaming_type = streaming["type"].strip().lower()
manifest = streaming["url"]
license_url = (
playback.get("fallback", {}).get("drm", {}).get("schemes", {}).get(self.drm_system, {}).get("licenseUrl")
or playback.get("drm", {}).get("schemes", {}).get(self.drm_system, {}).get("licenseUrl")
)
self.token = None
self.license = None
if streaming["protection"]["drmEnabled"]:
self.token = streaming["protection"]["drmToken"]
self.license = streaming["protection"]["schemes"]["widevine"]["licenseUrl"]
title.data["license_url"] = license_url
title.data["chapters"] = next((x.get("annotations") for x in playback["videos"] if x["type"] == "main"), None)
if streaming_type == "hls":
tracks = HLS.from_url(url=manifest, session=self.session).to_tracks(language=title.language)
dash = DASH.from_url(url=manifest, session=self.session)
tracks = dash.to_tracks(language="en", period_filter=self._period_filter)
elif streaming_type == "dash":
tracks = DASH.from_url(url=manifest, session=self.session).to_tracks(language=title.language)
for track in tracks:
track.is_original_lang = str(track.language) == original_language
track.name = "Original" if track.is_original_lang else track.name
else:
raise ValueError(f"Unknown streaming type: {streaming_type}")
if isinstance(track, Audio):
role = track.data["dash"]["representation"].find("Role")
if role is not None and role.get("value") in ["description", "alternative", "alternate"]:
track.descriptive = True
if isinstance(track, Subtitle):
tracks.subtitles.remove(track)
subtitles = self._process_subtitles(dash, original_language)
tracks.add(subtitles)
return tracks
def get_chapters(self, title: Union[Movie, Episode]) -> Chapters:
return Chapters()
def get_chapters(self, title: Movie | Episode) -> Chapters:
if not title.data.get("chapters"):
return Chapters()
chapters = []
for chapter in title.data["chapters"]:
if "recap" in chapter.get("secondaryType", "").lower():
chapters.append(Chapter(name="Recap", timestamp=chapter["start"]))
if chapter.get("end"):
chapters.append(Chapter(timestamp=chapter.get("end")))
if "intro" in chapter.get("secondaryType", "").lower():
chapters.append(Chapter(name="Intro", timestamp=chapter["start"]))
if chapter.get("end"):
chapters.append(Chapter(timestamp=chapter.get("end")))
elif "credits" in chapter.get("type", "").lower():
chapters.append(Chapter(name="Credits", timestamp=chapter["start"]))
if not any(c.timestamp == 0 for c in chapters):
chapters.append(Chapter(timestamp=0))
def get_widevine_service_certificate(self, **_: Any) -> str:
return None
return sorted(chapters, key=lambda x: x.timestamp)
def get_widevine_license(self, challenge: bytes, **_: Any) -> str:
if not self.license:
def get_widevine_service_certificate(self, challenge: bytes, title: Episode | Movie, **_: Any) -> str:
if not (license_url := title.data.get("license_url")):
return None
return self.session.post(url=license_url, data=challenge).content
def get_widevine_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
if not (license_url := title.data.get("license_url")):
return None
r = self.session.post(self.license, headers={"Preauthorization": self.token}, data=challenge)
if not r.ok:
raise ConnectionError(r.text)
r = self.session.post(url=license_url, data=challenge)
if r.status_code != 200:
raise ConnectionError(r.status_code, r.text)
return r.content
def get_playready_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
if not (license_url := title.data.get("license_url")):
return None
r = self.session.post(url=license_url, data=challenge)
if r.status_code != 200:
raise ConnectionError(r.status_code, r.text)
return r.content
# Service specific functions
@staticmethod
def _process_subtitles(dash: DASH, language: str) -> list[Subtitle]:
subtitle_groups = defaultdict(list)
manifest = dash.manifest
for period in manifest.findall("Period"):
for adapt_set in period.findall("AdaptationSet"):
if adapt_set.get("contentType") != "text" or not adapt_set.get("lang"):
continue
role = adapt_set.find("Role")
label = adapt_set.find("Label")
key = (
adapt_set.get("lang"),
role.get("value") if role is not None else "subtitle",
label.text if label is not None else "",
)
subtitle_groups[key].append((period, adapt_set))
final_tracks = []
for (lang, role_value, label_text), adapt_set_group in subtitle_groups.items():
first_period, first_adapt = adapt_set_group[0]
if first_adapt.find("Representation") is None:
continue
s_elements_with_context = []
for _, adapt_set in adapt_set_group:
rep = adapt_set.find("Representation")
if rep is None:
continue
template = rep.find("SegmentTemplate") or adapt_set.find("SegmentTemplate")
timeline = template.find("SegmentTimeline") if template is not None else None
if timeline is not None:
start_num = int(template.get("startNumber", 1))
s_elements_with_context.extend((start_num, s_elem) for s_elem in timeline.findall("S"))
s_elements_with_context.sort(key=lambda x: x[0])
combined_adapt = deepcopy(first_adapt)
combined_rep = combined_adapt.find("Representation")
seg_template = combined_rep.find("SegmentTemplate")
if seg_template is None:
template_at_adapt = combined_adapt.find("SegmentTemplate")
if template_at_adapt is not None:
seg_template = deepcopy(template_at_adapt)
combined_rep.append(seg_template)
combined_adapt.remove(template_at_adapt)
else:
continue
if seg_template.find("SegmentTimeline") is not None:
seg_template.remove(seg_template.find("SegmentTimeline"))
new_timeline = etree.Element("SegmentTimeline")
new_timeline.extend(deepcopy(s) for _, s in s_elements_with_context)
seg_template.append(new_timeline)
seg_template.set("startNumber", "1")
if "endNumber" in seg_template.attrib:
del seg_template.attrib["endNumber"]
track_id = hex(crc32(f"sub-{lang}-{role_value}-{label_text}".encode()) & 0xFFFFFFFF)[2:]
lang_obj = Language.get(lang)
track_name = "Original" if (language and is_close_match(lang_obj, [language])) else lang_obj.display_name()
final_tracks.append(
Subtitle(
id_=track_id,
url=dash.url,
codec=Subtitle.Codec.WebVTT,
language=lang_obj,
is_original_lang=bool(language and is_close_match(lang_obj, [language])),
descriptor=Track.Descriptor.DASH,
sdh="sdh" in label_text.lower() or role_value == "caption",
forced="forced" in label_text.lower() or "forced" in role_value.lower(),
name=track_name,
data={
"dash": {
"manifest": manifest,
"period": first_period,
"adaptation_set": combined_adapt,
"representation": combined_rep,
}
},
)
)
return final_tracks
@staticmethod
def _period_filter(period: Any) -> bool:
"""Shouldn't be needed for fallback manifest"""
if not (duration := period.get("duration")):
return False
return DASH.pt_to_sec(duration) < 120
def _show(self, title: str) -> Episode:
params = {
"include": "default",
"decorators": "playbackAllowed,contentAction,badges",
"decorators": "viewingHistory,badges,isFavorite,contentAction",
}
data = self._request("GET", "/cms/routes/show/{}".format(title), params=params)
content = next(x for x in data["included"] if x["attributes"].get("alias") == "generic-show-episodes")
content_id = content["id"]
show_id = content["attributes"]["component"]["mandatoryParams"]
info = next(x for x in data["included"] if x.get("attributes", {}).get("alternateId", "") == title)
content = next((x for x in data["included"] if "show-page-rail-episodes-tabbed-content" in x["attributes"].get("alias", "")), None)
if not content:
raise ValueError("Show not found")
content_id = content.get("id")
show_id = content["attributes"]["component"].get("mandatoryParams", "")
season_params = [x.get("parameter") for x in content["attributes"]["component"]["filters"][0]["options"]]
page = next(x for x in data["included"] if x.get("type", "") == "page")
seasons = [
self._request(
"GET", "/cms/collections/{}?{}&{}".format(content_id, season, show_id),
params={"include": "default", "decorators": "playbackAllowed,contentAction,badges"},
params={"include": "default", "decorators": "viewingHistory,badges,isFavorite,contentAction"},
)
for season in season_params
]
@@ -210,92 +472,112 @@ class DSCP(Service):
return [
Episode(
id_=ep["id"],
id_=ep["relationships"]["edit"]["data"]["id"],
service=self.__class__,
title=page["attributes"]["title"],
year=ep["attributes"]["airDate"][:4],
title=page["attributes"].get("title") or info["attributes"].get("originalName"),
year=ep["attributes"]["airDate"][:4] if ep["attributes"].get("airDate") else None,
season=ep["attributes"].get("seasonNumber"),
number=ep["attributes"].get("episodeNumber"),
name=ep["attributes"]["name"],
language=ep["attributes"]["audioTracks"][0]
if ep["attributes"].get("audioTracks")
else self.user_language,
data=ep,
)
for episodes in videos
for ep in episodes
if ep["attributes"]["videoType"] == "EPISODE"
if ep.get("attributes", {}).get("videoType", "") == "EPISODE"
]
def _episode(self, title: str) -> Episode:
params = {
"include": "default",
"decorators": "playbackAllowed,contentAction,badges",
}
data = self._request("GET", "/cms/routes/video/{}".format(title), params=params)
page = next((x for x in data["included"] if x.get("type", "") == "page"), None)
if not page:
raise IndexError("Episode page not found")
video_id = title.split("/")[1]
video_id = page["relationships"].get("primaryContent", {}).get("data", {}).get("id")
if not video_id:
raise IndexError("Episode id not found")
params = {"decorators": "isFavorite", "include": "primaryChannel"}
params = {"decorators": "isFavorite", "include": "show"}
content = self._request("GET", "/content/videos/{}".format(video_id), params=params)
episode = content["data"]["attributes"]
name = episode.get("name")
if episode.get("secondaryTitle"):
name += f" {episode.get('secondaryTitle')}"
episode = content.get("data", {}).get("attributes")
video_type = episode.get("videoType")
relationships = content.get("data", {}).get("relationships")
show = next((x for x in content["included"] if x.get("type", "") == "show"), {})
show_title = show.get("attributes", {}).get("name") or show.get("attributes", {}).get("originalName")
episode_name = episode.get("originalName") or episode.get("secondaryTitle")
if video_type.lower() in ("clip", "standalone_event"):
show_title = episode.get("originalName")
episode_name = episode.get("secondaryTitle", "")
return [
Episode(
id_=content["data"].get("id"),
id_=relationships.get("edit", {}).get("data", {}).get("id"),
service=self.__class__,
title=page["attributes"]["title"],
title=show_title,
year=int(episode.get("airDate")[:4]) if episode.get("airDate") else None,
season=episode.get("seasonNumber") or 0,
number=episode.get("episodeNumber") or 0,
name=name,
language=episode["audioTracks"][0] if episode.get("audioTracks") else self.user_language,
name=episode_name,
data=episode,
)
]
def _sport(self, title: str) -> Movie:
params = {
"include": "default",
"decorators": "viewingHistory,badges,isFavorite,contentAction",
}
data = self._request("GET", "/cms/routes/sport/{}".format(title), params=params)
def _request(
self,
method: str,
api: str,
params: dict = None,
headers: dict = None,
payload: dict = None,
) -> Any[dict | str]:
url = urljoin(self.base_url, api)
self.session.headers.update(self.config["headers"])
content = next((x for x in data["included"] if x.get("attributes", {}).get("alternateId", "") == title), None)
if not content:
raise ValueError(f"Content not found for title: {title}")
if params:
self.session.params.update(params)
if headers:
self.session.headers.update(headers)
movie = content.get("attributes")
relationships = content.get("relationships")
prep = self.session.prepare_request(Request(method, url, json=payload))
response = self.session.send(prep)
name = movie.get("name") or movie.get("originalName")
year = int(movie.get("firstAvailableDate")[:4]) if movie.get("firstAvailableDate") else None
return [
Movie(
id_=relationships.get("edit", {}).get("data", {}).get("id"),
service=self.__class__,
name=name + " - " + movie.get("secondaryTitle", ""),
year=year,
data=movie,
)
]
def _movie(self, title: str, entity: str) -> Movie:
params = {
"include": "default",
"decorators": "isFavorite,playbackAllowed,contentAction,badges",
}
data = self._request("GET", "/cms/routes/movie/{}".format(title), params=params)
movie = next((
x for x in data["included"]if x.get("attributes", {}).get("videoType", "").lower() == entity), None
)
if not movie:
raise ValueError("Movie not found")
return [
Movie(
id_=movie["relationships"]["edit"]["data"]["id"],
service=self.__class__,
name=movie["attributes"].get("name") or movie["attributes"].get("originalName"),
year=int(movie["attributes"]["airDate"][:4]) if movie["attributes"].get("airDate") else None,
data=movie,
)
]
def _request(self, method: str, endpoint: str, **kwargs: Any) -> Any[dict | str]:
url = urljoin(self.base_url, endpoint)
response = self.session.request(method, url, **kwargs)
try:
data = json.loads(response.content)
if data.get("errors"):
if "invalid.token" in data["errors"][0]["code"]:
self.log.error("- Invalid Token. Cookies are invalid or may have expired.")
sys.exit(1)
if "missingpackage" in data["errors"][0]["code"]:
self.log.error("- Access Denied. Title is not available for this subscription.")
sys.exit(1)
raise ConnectionError(data["errors"])
return data
except Exception as e:
raise ConnectionError("Request failed: {}".format(e))
raise ConnectionError(f"Request failed for {url}: {e}")
packages/envied/src/envied/services/DSCP/config.yaml
+4 -4
View File
@@ -1,4 +1,4 @@
headers:
user-agent: Chrome/96.0.4664.55
x-disco-client: WEB:UNKNOWN:dplus_us:2.44.4
x-disco-params: realm=go,siteLookupKey=dplus_us,bid=dplus,hn=www.discoveryplus.com,hth=,uat=false
endpoints:
base_url: "https://default.{}-{}.prd.api.discoveryplus.com"
client_id: "b6746ddc-7bc7-471f-a16c-f6aaf0c34d26" # androidtv
packages/envied/src/envied/services/NF/MSL/__init__.py
+1 -1
View File
@@ -26,7 +26,7 @@ from .schemes import EntityAuthenticationSchemes # noqa: F401
from .schemes import KeyExchangeSchemes
from .schemes.EntityAuthentication import EntityAuthentication
from .schemes.KeyExchangeRequest import KeyExchangeRequest
# from vinetrimmer.utils.wienvied.device import RemoteDevice
# from vinetrimmer.utils.widevine.device import RemoteDevice
class MSL:
log = logging.getLogger("MSL")
packages/envied/src/envied/services/NF/__init__.py
+287 -79
View File
@@ -20,6 +20,7 @@ from pymp4.parser import Box
from pywidevine import PSSH, Cdm
import requests
from langcodes import Language
from pathlib import Path
from envied.core.constants import AnyTrack
from envied.core.credential import Credential
@@ -29,7 +30,7 @@ from envied.core.titles import Titles_T, Title_T
from envied.core.titles.episode import Episode, Series
from envied.core.titles.movie import Movie, Movies
from envied.core.titles.title import Title
from envied.core.tracks import Tracks, Chapters
from envied.core.tracks import Tracks, Chapters, Hybrid
from envied.core.tracks.audio import Audio
from envied.core.tracks.chapter import Chapter
from envied.core.tracks.subtitle import Subtitle
@@ -76,7 +77,7 @@ class NF(Service):
@click.option("-hb", "--high-bitrate", is_flag=True, default=False, help="Get more video bitrate")
@click.pass_context
def cli(ctx, **kwargs):
return Netflix(ctx, **kwargs)
return NF(ctx, **kwargs)
def __init__(self, ctx: click.Context, title: str, drm_system: Literal["widevine", "playready"], profile: str, meta_lang: str, hydrate_track: bool, high_bitrate: bool):
super().__init__(ctx)
@@ -162,39 +163,74 @@ class NF(Service):
def get_tracks(self, title: Title_T) -> Tracks:
tracks = Tracks()
# If Video Codec is H.264 is selected but `self.profile is none` profile QC has to be requested seperately
def mark_repack(track_group):
# mark videos + audio
for t in track_group.videos + track_group.audio:
t.needs_repack = True
# mark subtitles
for t in getattr(track_group, "subtitles", []):
t.needs_repack = True
# -------------------------------
# Parse manifests / fetch tracks
# -------------------------------
if self.vcodec == Video.Codec.AVC:
# self.log.info(f"Profile: {self.profile}")
try:
manifest = self.get_manifest(title, self.profiles)
movie_track = self.manifest_as_tracks(manifest, title, self.hydrate_track)
mark_repack(movie_track)
tracks.add(movie_track)
if self.profile is not None:
self.log.info(f"Requested profiles: {self.profile}")
else:
qc_720_profile = [x for x in self.config["profiles"]["video"][self.vcodec.extension.upper()]["QC"] if "l40" not in x and 720 in self.quality]
qc_manifest = self.get_manifest(title, qc_720_profile if 720 in self.quality else self.config["profiles"]["video"][self.vcodec.extension.upper()]["QC"])
qc_720_profile = [
x for x in self.config["profiles"]["video"][self.vcodec.extension.upper()]["QC"]
if "l40" not in x and 720 in self.quality
]
# QC profiles
qc_manifest = self.get_manifest(
title,
qc_720_profile if 720 in self.quality
else self.config["profiles"]["video"][self.vcodec.extension.upper()]["QC"]
)
qc_tracks = self.manifest_as_tracks(qc_manifest, title, False)
mark_repack(qc_tracks)
tracks.add(qc_tracks.videos)
mpl_manifest = self.get_manifest(title, [x for x in self.config["profiles"]["video"][self.vcodec.extension.upper()]["MPL"] if "l40" not in x])
# MPL Profiles
mpl_manifest = self.get_manifest(
title,
[x for x in self.config["profiles"]["video"][self.vcodec.extension.upper()]["MPL"]
if "l40" not in x]
)
mpl_tracks = self.manifest_as_tracks(mpl_manifest, title, False)
mark_repack(mpl_tracks)
tracks.add(mpl_tracks.videos)
except Exception as e:
self.log.error(e)
else:
# HEVC / DV / HDR mode
if self.high_bitrate:
splitted_profiles = self.split_profiles(self.profiles)
for index, profile_list in enumerate(splitted_profiles):
try:
self.log.debug(f"Index: {index}. Getting profiles: {profile_list}")
manifest = self.get_manifest(title, profile_list)
manifest_tracks = self.manifest_as_tracks(manifest, title, self.hydrate_track if index == 0 else False)
manifest_tracks = self.manifest_as_tracks(
manifest,
title,
self.hydrate_track if index == 0 else False
)
mark_repack(manifest_tracks)
tracks.add(manifest_tracks if index == 0 else manifest_tracks.videos)
except Exception:
self.log.error(f"Error getting profile: {profile_list}. Skipping")
continue
@@ -202,49 +238,160 @@ class NF(Service):
try:
manifest = self.get_manifest(title, self.profiles)
manifest_tracks = self.manifest_as_tracks(manifest, title, self.hydrate_track)
mark_repack(manifest_tracks)
tracks.add(manifest_tracks)
except Exception as e:
self.log.error(e)
# --------------------------------------------------------
# 🧩 HYBRID DV+HDR Injection (copied from 1st script)
# --------------------------------------------------------
video_ranges = [v.range for v in tracks.videos]
has_dv = Video.Range.DV in video_ranges
has_hdr10 = Video.Range.HDR10 in video_ranges
has_hdr10p = Video.Range.HDR10P in video_ranges
if self.range[0] == Video.Range.HYBRID and has_hdr10 and (has_dv or has_hdr10p):
try:
self.log.info("Performing HYBRID DV+HDR injection...")
hdr_video = next((v for v in tracks.videos if v.range == Video.Range.HDR10), None)
dv_video = next((v for v in tracks.videos if v.range in (Video.Range.DV, Video.Range.HDR10P)), None)
if not hdr_video or not dv_video:
raise Exception("Missing HDR10 or DV video track for hybrid merge")
# Ensure both files exist before injection
def ensure_local_file(video):
if not getattr(video, "path", None) or not os.path.exists(video.path):
temp_path = config.directories.temp / f"{video.id}.hevc"
self.log.info(f"Downloading temporary stream for {video.range} → {temp_path.name}")
with self.session.get(video.url, stream=True) as r:
r.raise_for_status()
with open(temp_path, "wb") as f:
for chunk in r.iter_content(chunk_size=1024 * 1024):
f.write(chunk)
video.path = temp_path
return video.path
ensure_local_file(hdr_video)
ensure_local_file(dv_video)
# Perform hybrid merge
Hybrid([hdr_video, dv_video], self.__class__.__name__.lower())
injected_path = config.directories.temp / "HDR10-DV.hevc"
self.log.info(f"Hybrid file created → {injected_path}")
# Replace HDR10 with merged track
hdr_video.range = Video.Range.DV
hdr_video.path = injected_path
except Exception as e:
self.log.warning(f"Hybrid injection failed: {e}")
# --------------------------------------------------------
# Disable proxy for all tracks
# --------------------------------------------------------
for track in tracks:
track.needs_proxy = False
# --------------------------------------------------------
# Add Attachments + Save poster
# --------------------------------------------------------
try:
if isinstance(title, Movie):
poster_url = title.data["boxart"][0]["url"]
else:
poster_url = title.data["stills"][0]["url"]
# Temp directory
temp_dir = Path(self.config.get("directories", {}).get("Downloads", "./Downloads"))
temp_dir.mkdir(parents=True, exist_ok=True)
# Add Attachments for profile picture
if isinstance(title, Movie):
tracks.add(
Attachment.from_url(
url=title.data["boxart"][0]["url"]
)
)
else:
tracks.add(
Attachment.from_url(title.data["stills"][0]["url"])
)
poster_path = temp_dir / "poster.jpg"
# Save poster locally
try:
resp = requests.get(poster_url, timeout=15)
if resp.status_code == 200:
with open(poster_path, "wb") as f:
f.write(resp.content)
except Exception as e:
self.log.error(f"Failed to save poster.jpg: {e}")
# Create attachment
attachment = Attachment.from_url(url=poster_url)
attachment.filename = str(poster_path)
tracks.add(attachment)
except Exception as e:
self.log.error(f"Failed to add attachments: {e}")
return tracks
return tracks
def split_profiles(self, profiles: List[str]) -> List[List[str]]:
"""
Split profiles with names containing specific patterns based on video codec
For H264: uses patterns "l30", "l31", "l40" (lowercase)
For non-H264: uses patterns "L30", "L31", "L40", "L41", "L50", "L51" (uppercase)
Returns List[List[str]] type with profiles grouped by pattern
Split profiles based on codec level ranges and also DV/HDR groups for HYBRID mode.
"""
# Define the profile patterns to match based on video codec
if self.vcodec == Video.Codec.AVC: # H264
patterns = ["l30", "l31", "l40"]
# -----------------------------
# Patterns for profile splitting
# -----------------------------
if self.vcodec == Video.Codec.AVC:
level_patterns = ["l30", "l31", "l40"]
else:
patterns = ["L30", "L31", "L40", "L41", "L50", "L51"]
# Group profiles by pattern
level_patterns = ["L30", "L31", "L40", "L41", "L50", "L51"]
# -----------------------------
# HYBRID MODE — Add DV/HDR splits
# -----------------------------
dv_patterns = ["DV", "dv"]
hdr10_patterns = ["HDR10", "hdr10"]
hdr10p_patterns = ["HDR10P", "hdr10p"]
result: List[List[str]] = []
for pattern in patterns:
pattern_group = []
for profile in profiles:
if pattern in profile:
pattern_group.append(profile)
if pattern_group: # Only add non-empty groups
result.append(pattern_group)
used = set()
# -----------------------------
# Group DV profiles first
# -----------------------------
if self.range[0] == Video.Range.HYBRID:
dv_group = [p for p in profiles if any(tag in p for tag in dv_patterns)]
if dv_group:
result.append(dv_group)
used.update(dv_group)
# Group HDR10 profiles
hdr10_group = [p for p in profiles if any(tag in p for tag in hdr10_patterns)]
if hdr10_group:
result.append(hdr10_group)
used.update(hdr10_group)
# Group HDR10+ profiles
hdr10p_group = [p for p in profiles if any(tag in p for tag in hdr10p_patterns)]
if hdr10p_group:
result.append(hdr10p_group)
used.update(hdr10p_group)
# -----------------------------
# Normal HEVC/H264 Level Splitting
# -----------------------------
for pattern in level_patterns:
group = [p for p in profiles if (pattern in p and p not in used)]
if group:
result.append(group)
used.update(group)
# -----------------------------
# Any remaining profiles
# -----------------------------
leftover = [p for p in profiles if p not in used]
if leftover:
result.append(leftover)
return result
@@ -324,42 +471,61 @@ class NF(Service):
# return super().get_widevine_license(challenge=challenge, title=title, track=track)
def configure(self):
# self.log.info(ctx)
# if profile is none from argument let's use them all profile in video codec scope
# self.log.info(f"Requested profiles: {self.profile}")
# -----------------------------
# Profiles selection
# -----------------------------
if self.profile is None:
self.profiles = self.config["profiles"]["video"][self.vcodec.extension.upper()]
if self.profile is not None:
self.requested_profiles = self.profile.split('+')
self.log.info(f"Requested profile: {self.requested_profiles}")
else:
# self.log.info(f"Video Range: {self.range}")
self.requested_profiles = self.config["profiles"]["video"][self.vcodec.extension.upper()]
# Make sure video codec is supported by Netflix
# -----------------------------
# Validate codec support
# -----------------------------
if self.vcodec.extension.upper() not in self.config["profiles"]["video"]:
raise ValueError(f"Video Codec {self.vcodec} is not supported by Netflix")
if self.range[0].name not in list(self.config["profiles"]["video"][self.vcodec.extension.upper()].keys()) and self.vcodec != Video.Codec.AVC and self.vcodec != Video.Codec.VP9:
self.log.error(f"Video range {self.range[0].name} is not supported by Video Codec: {self.vcodec}")
sys.exit(1)
if len(self.range) > 1:
self.log.error(f"Multiple video range is not supported right now.")
sys.exit(1)
if self.vcodec == Video.Codec.AVC and self.range[0] != Video.Range.SDR:
self.log.error(f"H.264 Video Codec only supports SDR")
sys.exit(1)
# -----------------------------
# HYBRID MODE FIX
# -----------------------------
if self.range[0] == Video.Range.HYBRID:
# Only allowed for HEVC
if self.vcodec != Video.Codec.HEVC:
self.log.error("HYBRID mode is only supported for HEVC codec.")
sys.exit(1)
self.log.info("HYBRID mode detected → Skipping standard range validation")
# Skip all range validation completely
else:
# Normal validation path (non-HYBRID)
if self.range[0].name not in list(self.config["profiles"]["video"][self.vcodec.extension.upper()].keys()) \
and self.vcodec not in (Video.Codec.AVC, Video.Codec.VP9):
self.log.error(f"Video range {self.range[0].name} is not supported by Video Codec: {self.vcodec}")
sys.exit(1)
if len(self.range) > 1:
self.log.error("Multiple video range is not supported right now.")
sys.exit(1)
if self.vcodec == Video.Codec.AVC and self.range[0] != Video.Range.SDR:
self.log.error("H.264 Video Codec only supports SDR")
sys.exit(1)
# -----------------------------
# Final profile resolution
# -----------------------------
self.profiles = self.get_profiles()
self.log.info("Intializing a MSL client")
self.log.info("Initializing a MSL client")
self.get_esn()
scheme = KeyExchangeSchemes.AsymmetricWrapped
self.log.info(f"Scheme: {scheme}")
self.msl = MSL.handshake(
scheme=scheme,
session=self.session,
@@ -367,6 +533,7 @@ class NF(Service):
sender=self.esn.data,
cache=self.cache.get("MSL")
)
cookie = self.session.cookies.get_dict()
self.userauthdata = UserAuthentication.NetflixIDCookies(
netflixid=cookie["NetflixId"],
@@ -376,25 +543,66 @@ class NF(Service):
def get_profiles(self):
result_profiles = []
# -------------------------------
# AVC logic (unchanged)
# -------------------------------
if self.vcodec == Video.Codec.AVC:
if self.requested_profiles is not None:
for requested_profiles in self.requested_profiles:
result_profiles.extend(flatten(list(self.config["profiles"]["video"][self.vcodec.extension.upper()][requested_profiles])))
for req in self.requested_profiles:
result_profiles.extend(
flatten(list(self.config["profiles"]["video"][self.vcodec.extension.upper()][req]))
)
return result_profiles
result_profiles.extend(flatten(list(self.config["profiles"]["video"][self.vcodec.extension.upper()].values())))
result_profiles.extend(
flatten(list(self.config["profiles"]["video"][self.vcodec.extension.upper()].values()))
)
return result_profiles
# Handle case for codec VP9
# -------------------------------
# VP9 logic (unchanged)
# -------------------------------
if self.vcodec == Video.Codec.VP9 and self.range[0] != Video.Range.HDR10:
result_profiles.extend(self.config["profiles"]["video"][self.vcodec.extension.upper()].values())
result_profiles.extend(
self.config["profiles"]["video"][self.vcodec.extension.upper()].values()
)
return result_profiles
# -------------------------------
# HEVC Hybrid mode (FIXED)
# -------------------------------
if self.vcodec == Video.Codec.HEVC and self.range[0] == Video.Range.HYBRID:
self.log.info("HYBRID mode detected → Using HDR10 + DV profiles")
hevc_profiles = self.config["profiles"]["video"][self.vcodec.extension.upper()]
result_profiles = []
# 1. HDR10 FIRST
if "HDR10" in hevc_profiles:
result_profiles += hevc_profiles["HDR10"]
# 2. HDR10P (some titles use this instead of HDR10)
if "HDR10P" in hevc_profiles:
result_profiles += hevc_profiles["HDR10P"]
# 3. DV LAST (IMPORTANT!)
if "DV" in hevc_profiles:
result_profiles += hevc_profiles["DV"]
return result_profiles
# -------------------------------
# Normal HEVC (non HYBRID)
# -------------------------------
for profiles in self.config["profiles"]["video"][self.vcodec.extension.upper()]:
for range in self.range:
if range in profiles:
result_profiles.extend(self.config["profiles"]["video"][self.vcodec.extension.upper()][range.name])
# sys.exit(1)
for r in self.range:
if r in profiles:
result_profiles.extend(
self.config["profiles"]["video"][self.vcodec.extension.upper()][r.name]
)
self.log.debug(f"Result_profiles: {result_profiles}")
return result_profiles
@@ -767,4 +975,4 @@ class NF(Service):
return Video.Range.SDR
# If profile not found, return SDR as default
return Video.Range.SDR
return Video.Range.SDR
packages/envied/src/envied/services/RKTN/__init__.py
+791
View File
@@ -0,0 +1,791 @@
import base64
from copy import copy
import datetime
import hashlib
import hmac
import json
import re
import os
import sys
from typing import Optional
from aiohttp import CookieJar
from pymediainfo import MediaInfo
from langcodes import Language
import click
import urllib.parse
from requests import HTTPError
from envied.core.config import config
from envied.core.constants import AnyTrack
from envied.core.credential import Credential
from envied.core.manifests.dash import DASH
from envied.core.service import Service
from envied.core.titles import Title_T
from envied.core.titles.episode import Episode, Series
from envied.core.titles.movie import Movie, Movies
from envied.core.tracks.audio import Audio
from envied.core.tracks.chapters import Chapters
from envied.core.tracks.subtitle import Subtitle
from envied.core.tracks.tracks import Tracks
from envied.core.tracks.video import Video
from pyplayready.cdm import Cdm as PlayReadyCdm
class RKTN(Service):
"""
Service code for Rakuten's Rakuten TV streaming service (https://rakuten.tv).
\b
Authorization: Credentials
Security: FHD-UHD@L1, SD-FHD@L3; with trick
\b
Maximum of 3 audio tracks, otherwise will fail because Rakuten blocks more than 3 requests.
Subtitles requests expires fast, so together with video and audio it will fail.
If you want subs, use -S or -na -nv -nc, and download the rest separately.
\b
Command for Titles with no SDR (if not set range to HDR10 it will fail):
uv run unshackle dl -r HDR10 [OPTIONS] RKTN -m https://www.rakuten.tv/...
\b
TODO: - TV Shows are not yet supported as there's 0 TV Shows to purchase, rent, or watch in my region
\b
NOTES: - Only movies are supported as my region's Rakuten has no TV shows available to purchase at all
"""
ALIASES = ["RakutenTV", "rakuten", "rakutentv"]
TITLE_RE = r"^(?:https?://(?:www\.)?rakuten\.tv/([a-z]+/|)movies(?:/[a-z]{2})?/)(?P<id>[a-z0-9-]+)"
LANG_MAP = {
"es": "es-ES",
"pt": "pt-PT",
}
@staticmethod
@click.command(name="RakutenTV", short_help="https://rakuten.tv")
@click.argument("title", type=str, required=False)
@click.option(
"-dev",
"--device",
default=None,
type=click.Choice(
[
"web", # Device: Web Browser - Maximum Quality: 720p - DRM: Widevine
"android", # Device: Android Phone - Maximum Quality: 720p - DRM: Widevine
"atvui40", # Device: AndroidTV - Maximum Quality: 2160p - DRM: Widevine
"lgui40", # Device: LG SMART TV - Maximum Quality: 2160p - DRM: Playready
"smui40", # Device: Samsung SMART TV - Maximum Quality: 2160p - DRM: Playready
],
case_sensitive=True,
),
help="The device you want to make requests with.",
)
@click.option(
"-m", "--movie", is_flag=True, default=False, help="Title is a movie."
)
@click.option(
"-dal", "--desired-audio-language", type=str, default="SPA,ENG", help="Select desired audio language tracks for this title. Default SPA,ENG. Separate multiple languages with a comma."
)
@click.pass_context
def cli(ctx, **kwargs):
return RKTN(ctx, **kwargs)
def __init__(self, ctx, title, device, movie, desired_audio_language):
super().__init__(ctx)
#self.parse_title(ctx, title)
self.title = title
self.cdm = ctx.obj.cdm
self.playready = isinstance(self.cdm, PlayReadyCdm)
self.desired_audio_language = desired_audio_language
self.range = ctx.parent.params.get("range_")[0].name or "SDR"
self.vcodec = ctx.parent.params.get("vcodec") or Video.Codec.AVC # Defaults to H264
self.resolution = "UHD" if (self.vcodec.extension.lower() == "h265" or self.range in ['HYBRID', 'HDR10', 'HDR10P', 'DV']) else "FHD"
self.device = "lgui40" if self.playready else "android"
self.movie = movie or "movies" in title
self.audio_languages = []
# set a custom device if provided
if device is not None:
self.device = device
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential)
if not credential:
raise EnvironmentError("Service requires Credentials for Authentication.")
self.session.headers.update(
{
"Origin": "https://rakuten.tv/",
"User-Agent": "Mozilla/5.0 (Linux; Android 11; SHIELD Android TV Build/RQ1A.210105.003; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/99.0.4844.88 Mobile Safari/537.36",
}
)
def get_titles(self):
self.pair_device()
if self.movie:
endpoint = self.config["endpoints"]["title"]
else:
endpoint = self.config["endpoints"]["show"]
params = urllib.parse.urlencode(
{
"classification_id": self.classification_id,
"device_identifier": self.config["clients"][self.device][
"device_identifier"
],
"device_serial": self.config["clients"][self.device]["device_serial"],
"locale": self.locale,
"market_code": self.market_code,
"session_uuid": self.session_uuid,
"timestamp": f"{int(datetime.datetime.now().timestamp())}005",
"support_closed_captions": "true",
}
)
title_url = endpoint.format(
title_id=self.title
) + params
title = self.session.get(url=title_url).json()
if "errors" in title:
error = title["errors"][0]
if error["code"] == "error.not_found":
self.log.error(f"Title [{self.title}] was not found on this account.")
else:
self.log.error(
f"Unable to get title info: {error['message']} [{error['code']}]"
)
sys.exit(1)
title = self.get_info(title["data"])
if self.movie:
return Movies(
[
Movie(
id_=self.title,
service=self.__class__,
name=title["title"],
year=title["year"],
language="en",
data=title,
description=title["plot"],
)
]
)
else:
episodes_list = []
#title_ep = self.get_info(title["data"]['episodes'])
for season in title["tv_show"]["seasons"]:
data_season = endpoint.format(
title_id=season["id"]
) + params
data = self.session.get(url=data_season).json()
if "errors" in data:
error = data["errors"][0]
if error["code"] == "error.not_found":
self.log.error(f"Season [{season['id']}] was not found on this account.")
else:
self.log.error(
f"Unable to get title info: {error['message']} [{error['code']}]"
)
continue
for episode in data["data"]["episodes"]:
episodes_list.append(
Episode(
id_=episode["id"],
service=self.__class__,
title=episode["tv_show_title"],
season=episode["season_number"],
number=episode["number"],
name=episode["title"] or episode['display_name'],
description=episode["short_plot"],
year=episode["year"],
language="en",
data=episode,
)
)
return Series(episodes_list)
def get_tracks(self, title: Title_T) -> Tracks:
# Obtener tracks para todos los idiomas de audio disponibles
all_tracks = None
for audio_lang in self.audio_languages:
self.log.info(f"Getting tracks for audio language: {audio_lang}")
# Obtener stream info para este idioma específico
stream_info = self.get_avod(audio_lang, title) if self.kind == "avod" else self.get_me(audio_lang, title)
if "errors" in stream_info:
error = stream_info["errors"][0]
if "error.streaming.no_active_right" in stream_info["errors"][0]["code"]:
self.log.error(
" x You don't have the rights for this content\n You need to rent or buy it first"
)
else:
self.log.error(
f" - Failed to get track info: {error['message']} [{error['code']}]"
)
sys.exit(1)
stream_info = stream_info["data"]["stream_infos"][0]
if all_tracks is None:
# Primera iteración: crear el objeto tracks principal
self.license_url = stream_info["license_url"]
all_tracks = DASH.from_url(url=stream_info["url"], session=self.session).to_tracks(language=title.language)
# Procesar subtítulos (solo una vez)
subtitle_tracks = []
for subtitle in stream_info.get("all_subtitles", []):
subtitle_tracks += [
Subtitle(
id_=hashlib.md5(subtitle["url"].encode()).hexdigest()[0:6],
url=subtitle["url"],
codec=Subtitle.Codec.from_mime(subtitle["format"]),
forced=subtitle["forced"],
language=subtitle["locale"],
)
]
all_tracks.add(subtitle_tracks)
else:
# Iteraciones adicionales: obtener tracks de audio adicionales
temp_tracks = DASH.from_url(url=stream_info["url"], session=self.session).to_tracks(language=title.language)
# Agregar solo los tracks de audio nuevos
for audio_track in temp_tracks.audio:
# Verificar que no sea duplicado basado en el idioma y codec
is_duplicate = False
for existing_audio in all_tracks.audio:
if (existing_audio.language == audio_track.language and
existing_audio.codec == audio_track.codec):
is_duplicate = True
break
if not is_duplicate:
all_tracks.audio.append(audio_track)
# Procesar HDR para videos
for video in all_tracks.videos:
if "HDR10" in video.url:
video.range = Video.Range.HDR10
# Aplicar el método append_tracks mejorado
self.append_tracks(all_tracks)
return all_tracks
def get_chapters(self, title: Title_T) -> Chapters:
return Chapters([])
def get_me(self, audio_language=None, title: Title_T = None):
# Si no se especifica idioma, usar el primero disponible
if audio_language is None:
audio_language = self.audio_languages[0]
stream_info_url = self.config["endpoints"]["manifest"].format(
kind="me"
) + urllib.parse.urlencode(
{
"audio_language": audio_language, # Usar el idioma especificado
"audio_quality": "5.1", # Will get better audio in different request to make sure it wont error
"classification_id": self.classification_id,
"content_id": title.id,
"content_type": "movies" if self.movie else "episodes",
"device_identifier": self.config["clients"][self.device][
"device_identifier"
],
"device_serial": "not_implemented",
"device_stream_audio_quality": "5.1",
"device_stream_hdr_type": self.hdr_type,
"device_stream_video_quality": self.resolution,
"device_uid": "affa434b-8b7c-4ff3-a15e-df1fe500e71e",
"device_year": self.config["clients"][self.device]["device_year"],
"disable_dash_legacy_packages": "false",
"gdpr_consent": self.config["gdpr_consent"],
"gdpr_consent_opt_out": 0,
"hdr_type": self.hdr_type,
"ifa_subscriber_id": self.ifa_subscriber_id,
"locale": self.locale,
"market_code": self.market_code,
"player": self.config["clients"][self.device]["player"],
"player_height": 1080,
"player_width": 1920,
"publisher_provided_id": "046f58b1-d89b-4fa4-979b-a9bcd6d78a76",
"session_uuid": self.session_uuid,
"strict_video_quality": "false",
"subtitle_formats": ["vtt"],
"subtitle_language": "MIS",
"timestamp": f"{int(datetime.datetime.now().timestamp())}122",
"video_type": "stream",
}
)
stream_info_url += "&signature=" + self.generate_signature(stream_info_url)
return self.session.post(
url=stream_info_url,
).json()
def get_avod(self, audio_language=None, title: Title_T = None):
# Si no se especifica idioma, usar el primero disponible
if audio_language is None:
audio_language = self.audio_languages[0]
stream_info_url = self.config["endpoints"]["manifest"].format(
kind="avod"
) + urllib.parse.urlencode(
{
"device_stream_video_quality": self.resolution,
"device_identifier": self.config["clients"][self.device][
"device_identifier"
],
"market_code": self.market_code,
"session_uuid": self.session_uuid,
"timestamp": f"{int(datetime.datetime.now().timestamp())}122",
}
)
stream_info_url += "&signature=" + self.generate_signature(stream_info_url)
return self.session.post(
url=stream_info_url,
data={
"hdr_type": self.hdr_type,
"audio_quality": "5.1", # Will get better audio in different request to make sure it wont error
"app_version": self.config["clients"][self.device]["app_version"],
"content_id": title.id,
"video_quality": self.resolution,
"audio_language": audio_language, # Usar el idioma especificado
"video_type": "stream",
"device_serial": self.config["clients"][self.device]["device_serial"],
"content_type": "movies" if self.movie else "episodes",
"classification_id": self.classification_id,
"subtitle_language": "MIS",
"player": self.config["clients"][self.device]["player"],
},
).json()
def generate_signature(self, url):
up = urllib.parse.urlparse(url)
digester = hmac.new(
self.access_token.encode(),
f"POST{up.path}{up.query}".encode(),
hashlib.sha1,
)
return (
base64.b64encode(digester.digest())
.decode("utf-8")
.replace("+", "-")
.replace("/", "_")
)
def append_tracks(self, tracks):
"""
Busca y agrega tracks adicionales de video y audio que no están en el manifest.
"""
if not tracks.videos:
self.log.warning("No video tracks found, skipping append_tracks")
return
# Buscar tracks de video adicionales
self._append_video_tracks(tracks)
# Buscar tracks de audio adicionales
self._append_audio_tracks(tracks)
def _append_video_tracks(self, tracks):
"""Busca y agrega tracks de video adicionales para H.264."""
if not tracks.videos:
return
codec = tracks.videos[0].codec
# Solo buscar tracks adicionales para H.264
if codec != Video.Codec.AVC:
self.log.debug(f"Skipping video track search (codec: {codec.name}, only works for AVC/H.264)")
return
# Extraer el patrón del codec de la URL
url_pattern = tracks.videos[-1].url
codec_match = re.search(r'(avc1|h264)-(\d+)', url_pattern, re.IGNORECASE)
if not codec_match:
self.log.debug("Could not find codec pattern in URL for video track search")
return
codec_prefix = codec_match.group(1) # "avc1" o "h264"
self.log.info(f"Searching for additional H.264 video tracks (pattern: {codec_prefix})...")
# Usar el directorio temp de Unshackle
temp_file = os.path.join(str(config.directories.temp), "video_test.mp4")
tracks_found = 0
for n in range(100):
# Generar URL del siguiente track
current_number = len(tracks.videos) + 1
ismv = re.sub(
rf"{codec_prefix}-\d+",
rf"{codec_prefix}-{current_number}",
tracks.videos[-1].url,
)
# Verificar si existe
try:
response = self.session.head(ismv, timeout=5)
if response.status_code != 200:
self.log.debug(f"Video track search ended at index {current_number}")
break
except Exception as e:
self.log.debug(f"Video track search failed: {e}")
break
# Crear copia del último video track
video = copy(tracks.videos[-1])
video.url = ismv
video.id_ = hashlib.md5(ismv.encode()).hexdigest()[:16]
# Descargar chunk para obtener info con MediaInfo
try:
with open(temp_file, "wb") as chunkfile:
data = self.session.get(
url=ismv,
headers={"Range": "bytes=0-50000"},
timeout=10
)
chunkfile.write(data.content)
# Parsear con MediaInfo
info = MediaInfo.parse(temp_file)
if not info.video_tracks:
self.log.debug(f"No video info found for track {current_number}")
continue
video_info = info.video_tracks[0]
video.height = video_info.height
video.width = video_info.width
video.bitrate = video_info.maximum_bit_rate or video_info.bit_rate
# Agregar el track
tracks.videos.append(video)
tracks_found += 1
self.log.info(
f" + Added video track #{current_number}: "
f"{video.width}x{video.height} @ {video.bitrate} bps"
)
except Exception as e:
self.log.warning(f"Failed to process video track {current_number}: {e}")
break
finally:
if os.path.exists(temp_file):
os.remove(temp_file)
if tracks_found > 0:
self.log.info(f"Total additional video tracks found: {tracks_found}")
def _append_audio_tracks(self, tracks):
"""Busca y agrega tracks de audio adicionales para todos los idiomas seleccionados."""
if not tracks.audio:
self.log.warning("No audio tracks found to use as base")
return
if not hasattr(self, 'audio_languages') or not self.audio_languages:
self.log.debug("No audio languages configured")
return
self.log.info(f"Searching for additional audio tracks in languages: {self.audio_languages}")
# Codecs a probar (en orden de preferencia)
codecs_to_try = ["ec-3", "ac-3", "dts", "mp4a"]
# Usar el directorio temp de Unshackle
temp_file = os.path.join(str(config.directories.temp), "audio_test.mp4")
base_audio = tracks.audio[0]
base_url = base_audio.url
tracks_found = 0
for language in self.audio_languages:
for codec_name in codecs_to_try:
# Generar URL del track
# Patrón: audio-{LANG}-{CODEC}-{NUMBER}
isma = re.sub(
r"audio-[a-zA-Z]{2,3}-[a-z0-9\-]+-\d+",
f"audio-{language.lower()}-{codec_name}-1",
base_url,
)
# Verificar si existe
try:
response = self.session.head(isma, timeout=5)
if response.status_code != 200:
continue
except Exception:
continue
# Verificar si ya existe (evitar duplicados)
if any(audio.url == isma for audio in tracks.audio):
self.log.debug(f"Audio track already exists: {language}-{codec_name}")
continue
# Crear nuevo track de audio
audio = copy(base_audio)
audio.url = isma
audio.id_ = hashlib.md5(isma.encode()).hexdigest()[:16]
# Mapear idioma
mapped_lang = self.LANG_MAP.get(language, language)
audio.language = Language.get(mapped_lang)
# Determinar si es idioma original
if tracks.videos:
audio.is_original_lang = (
audio.language.language == tracks.videos[0].language.language
)
# Obtener información del track con MediaInfo
try:
with open(temp_file, "wb") as bytetest:
data = self.session.get(
url=isma,
headers={"Range": "bytes=0-50000"},
timeout=10
)
bytetest.write(data.content)
info = MediaInfo.parse(temp_file)
if not info.audio_tracks:
self.log.debug(f"No audio info found for {language}-{codec_name}")
continue
audio_info = info.audio_tracks[0]
audio.bitrate = audio_info.bit_rate
# Detectar canales basado en codec
if codec_name in ["ec-3", "ac-3", "dts"]:
audio.channels = audio_info.channel_s or "5.1"
else: # mp4a (AAC)
audio.channels = audio_info.channel_s or "2.0"
# Actualizar codec
# Para Unshackle, necesitas mantener el formato correcto
audio.codec = Audio.Codec.from_codecs(codec_name)
# Agregar el track
tracks.audio.append(audio)
tracks_found += 1
self.log.info(
f" + Added audio track: {audio.language.display_name()} "
f"[{codec_name.upper()}] - {audio.channels}ch @ {audio.bitrate} bps"
)
except Exception as e:
self.log.debug(f"Failed to process audio {language}-{codec_name}: {e}")
finally:
if os.path.exists(temp_file):
os.remove(temp_file)
if tracks_found > 0:
self.log.info(f"Total additional audio tracks found: {tracks_found}")
def get_widevine_service_certificate(self, **kwargs):
return self.config["certificate"]
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]:
res = self.session.post(
url=self.license_url,
data=challenge,
)
if "errors" in res.text:
res = res.json()
if res["errors"][0]["message"] == "HttpException: Forbidden":
self.log.error(
" x This CDM is not eligible to decrypt this\n"
" content or has been blacklisted by RakutenTV"
)
elif res["errors"][0]["message"] == "HttpException: An error happened":
self.log.error(
" x This CDM seems to be revoked and\n"
" therefore it can't decrypt this content",
)
sys.exit(1)
return res.content
def get_playready_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]:
res = self.session.post(
url=self.license_url,
data=challenge,
)
if "errors" in res.text:
res = res.json()
if res["errors"][0]["message"] == "HttpException: Forbidden":
self.log.error(
" x This CDM is not eligible to decrypt this\n"
" content or has been blacklisted by RakutenTV"
)
elif res["errors"][0]["message"] == "HttpException: An error happened":
self.log.error(
" x This CDM seems to be revoked and\n"
" therefore it can't decrypt this content",
)
sys.exit(1)
return res.content
def pair_device(self):
# TODO: Make this return the tokens, move print out of the func
# log.info_("Logging into RakutenTV as an Android device")
if not self.credential:
self.log.error(" - No credentials provided, unable to log in.")
sys.exit(1)
try:
res = self.session.post(
url=self.config["endpoints"]["auth"],
params={
"device_identifier": self.config["clients"][self.device][
"device_identifier"
]
},
data={
"app_version": self.config["clients"][self.device]["app_version"],
"device_metadata[uid]": self.config["clients"][self.device][
"device_serial"
],
"device_metadata[os]": self.config["clients"][self.device][
"device_os"
],
"device_metadata[model]": self.config["clients"][self.device][
"device_model"
],
"device_metadata[year]": self.config["clients"][self.device][
"device_year"
],
"device_serial": self.config["clients"][self.device][
"device_serial"
],
"device_metadata[trusted_uid]": False,
"device_metadata[brand]": self.config["clients"][self.device][
"device_brand"
],
"classification_id": 69,
"user[password]": self.credential.password,
"device_metadata[app_version]": self.config["clients"][self.device][
"app_version"
],
"user[username]": self.credential.username,
"device_metadata[serial_number]": self.config["clients"][
self.device
]["device_serial"],
},
).json()
except HTTPError as e:
if e.response.status_code == 403:
self.log.error(
" - Rakuten returned a 403 (FORBIDDEN) error. "
"This could be caused by your IP being detected as a proxy, or regional issues. Cannot continue."
)
if "errors" in res:
error = res["errors"][0]
if "exception.forbidden_vpn" in error["code"]:
self.log.error(" x RakutenTV is detecting this VPN or Proxy")
else:
self.log.error(f" - Login failed: {error['message']} [{error['code']}]")
self.access_token = res["data"]["user"]["access_token"]
self.ifa_subscriber_id = res["data"]["user"]["avod_profile"][
"ifa_subscriber_id"
]
self.session_uuid = res["data"]["user"]["session_uuid"]
self.classification_id = res["data"]["user"]["profile"]["classification"]["id"]
self.locale = res["data"]["market"]["locale"]
self.market_code = res["data"]["market"]["code"]
def get_info(self, title):
self.kind = title["labels"]["purchase_types"][0]["kind"]
# self.available_resolutions = [x for x in title["labels"]["video_qualities"]]
# if any(x["abbr"] == "UHD" for x in title["labels"]["video_qualities"]):
# self.resolution = "UHD"
# elif any(x["abbr"] == "FHD" for x in title["labels"]["video_qualities"]):
# self.resolution = "FHD"
# elif any(x["abbr"] == "HD" for x in title["labels"]["video_qualities"]):
# self.resolution = "HD"
# else:
# self.resolution = "SD"
self.available_hdr_types = [x for x in title["labels"]["hdr_types"]]
if any(x["abbr"] == "HDR10_PLUS" for x in self.available_hdr_types) and any(
x["abbr"] == "HDR10_PLUS"
for x in title["view_options"]["support"]["hdr_types"]
):
self.hdr_type = "HDR10_PLUS"
elif any(x["abbr"] == "DOLBY_VISION" for x in self.available_hdr_types) and any(
x["abbr"] == "DOLBY_VISION"
for x in title["view_options"]["support"]["hdr_types"]
):
self.hdr_type = "DOLBY_VISION"
elif any(x["abbr"] == "HDR10" for x in self.available_hdr_types) and any(
x["abbr"] == "HDR10" for x in title["view_options"]["support"]["hdr_types"]
):
self.hdr_type = "HDR10"
else:
self.hdr_type = "NONE"
# Obtener view_options desde title o episodes
view_options = title.get("episodes", [{}])[0].get("view_options") or title.get("view_options")
# FIJO: Obtener TODOS los idiomas de audio disponibles
if len(view_options["private"]["offline_streams"]) == 1:
# Caso 1: Un solo stream con múltiples idiomas
self.audio_languages = [
x["abbr"]
for x in view_options["private"]["streams"][0]["audio_languages"]
]
else:
# Caso 2: Múltiples streams, obtener todos los idiomas únicos
all_audio_languages = []
for stream in view_options["private"]["streams"]:
for audio_lang in stream["audio_languages"]:
if audio_lang["abbr"] not in all_audio_languages:
all_audio_languages.append(audio_lang["abbr"])
self.audio_languages = all_audio_languages
# # TODO: Look up only for languages chosen by the user
# print(f"\nAvailable audio languages: {', '.join(self.audio_languages)}")
# selected = input("Type your desired languages, maximum of 3, UPPER CASE (ex: ENG,SPA,FRA): ")
selected_langs = [lang.strip() for lang in self.desired_audio_language.split(",") if lang.strip() in self.audio_languages]
if not selected_langs:
self.log.error("No selected language. Exiting.")
self.audio_languages = selected_langs
# Log para debug
self.log.info(f"Selected audio languages: {self.audio_languages}")
return title
packages/envied/src/envied/services/RKTN/config.yaml
+71
View File
@@ -0,0 +1,71 @@
certificate: |
CAUSxwUKwQIIAxIQFwW5F8wSBIaLBjM6L3cqjBiCtIKSBSKOAjCCAQoCggEBAJntWzsyfateJO/DtiqVtZhSCtW8yzdQPgZFuBTYdrjfQFEEQa2M462xG
7iMTnJaXkqeB5UpHVhYQCOn4a8OOKkSeTkwCGELbxWMh4x+Ib/7/up34QGeHleB6KRfRiY9FOYOgFioYHrc4E+shFexN6jWfM3rM3BdmDoh+07svUoQyk
dJDKR+ql1DghjduvHK3jOS8T1v+2RC/THhv0CwxgTRxLpMlSCkv5fuvWCSmvzu9Vu69WTi0Ods18Vcc6CCuZYSC4NZ7c4kcHCCaA1vZ8bYLErF8xNEkKd
O7DevSy8BDFnoKEPiWC8La59dsPxebt9k+9MItHEbzxJQAZyfWgkCAwEAAToUbGljZW5zZS53aWRldmluZS5jb20SgAOuNHMUtag1KX8nE4j7e7jLUnfS
SYI83dHaMLkzOVEes8y96gS5RLknwSE0bv296snUE5F+bsF2oQQ4RgpQO8GVK5uk5M4PxL/CCpgIqq9L/NGcHc/N9XTMrCjRtBBBbPneiAQwHL2zNMr80
NQJeEI6ZC5UYT3wr8+WykqSSdhV5Cs6cD7xdn9qm9Nta/gr52u/DLpP3lnSq8x2/rZCR7hcQx+8pSJmthn8NpeVQ/ypy727+voOGlXnVaPHvOZV+WRvWC
q5z3CqCLl5+Gf2Ogsrf9s2LFvE7NVV2FvKqcWTw4PIV9Sdqrd+QLeFHd/SSZiAjjWyWOddeOrAyhb3BHMEwg2T7eTo/xxvF+YkPj89qPwXCYcOxF+6gjo
mPwzvofcJOxkJkoMmMzcFBDopvab5tDQsyN9UPLGhGC98X/8z8QSQ+spbJTYLdgFenFoGq47gLwDS6NWYYQSqzE3Udf2W7pzk4ybyG4PHBYV3s4cyzdq8amvtE/sNSdOKReuHpfQ=
gdpr_consent: |
CPGeIEAPV65UAADABBNLCGCsAP_AAH_AAAAAHrsXZCpcBSlgYCpoAIoAKIAUEAAAgyAAABAAAoABCAAAIAQAgAAgIAAAAAAAAAAAIAJAAQAAAAEAAAAAAA
AAAAAIIACAAAAAIABAAAAAAAAACAAAAAAAAAAAAAAEAAAAgABAABAAAAAAEAAAAAAAAAAAAAAAAAAAAAAAAAgZ8xdkKlwFKWBgKGgAigAogBQQAACDIAAA
EAACAAAIAAAgBACAACAAAAAAAAAAAAAgAgABAAAAAQAAAAAAAAAAAAggAAAAAAAgAEAAAAAAAAAAAAAAAAAAAAAAAAQAAACAAEAAEAAAAAAQAA.YAAAAAAAA8DA
endpoints:
title: https://gizmo.rakuten.tv/v3/movies/{title_id}?
show: https://gizmo.rakuten.tv/v3/seasons/{title_id}?
manifest: https://gizmo.rakuten.tv/v3/{kind}/streamings?
auth: https://gizmo.rakuten.tv/v3/me/login_or_wuaki_link
clients:
web:
app_version: v3.0.11
device_identifier: web
device_serial: 6cc3584a-c182-4cc1-9f8d-b90e4ed76de9
player: web:DASH-CENC:WVM
device_os: Windows 10
device_model: GENERIC
device_year: 2019
device_brand: chrome
device_sdk: 100.0.4896
android:
app_version: 3.22.0
device_identifier: android
device_serial: 3187ad6c-4d1c-4cbb-9c59-8396d054eb2a
player: android:DASH-CENC
device_os: Android
device_model: SM-A105FN
device_year: 2021
device_brand: Samsung
device_sdk: ""
atvui40:
app_version: v2.77.0
device_identifier: atvui40
device_serial: 0424814603535001d1b1
player: atvui40:DASH-CENC:WVM
device_os: Android TV UI 40
device_model: SHIELD Android TV
device_year: 1970
device_brand: NVIDIA
device_sdk: ""
lgui40:
app_version: v2.77.0
device_identifier: lgui40
device_serial: 203WRMD8U920
player: lgui40:DASH-CENC:PR
device_os: LG UI 40
device_model: OLED65C11LB
device_year: 2021
device_brand: LG
device_sdk: ""
smui40:
app_version: v2.77.0
device_identifier: smui40
device_serial: 6cc3584a-c182-4cc1-9f8d-b90e4ed76de9
player: smtvui40:DASH-CENC:WVM
device_os: Samsung UI 40
device_model: QE43Q60RATXXH
device_year: 2019
device_brand: Samsung
device_sdk: ""
packages/envied/src/envied/services/TUBI/__init__.py
+80 -22
View File
@@ -6,16 +6,19 @@ import re
import sys
import uuid
from collections.abc import Generator
from typing import Any
from http.cookiejar import MozillaCookieJar
from typing import Any, Optional
import click
from langcodes import Language
from pyplayready.cdm import Cdm as PlayReadyCdm
from envied.core.credential import Credential
from envied.core.downloaders import aria2c, requests
from envied.core.manifests import DASH
from envied.core.search_result import SearchResult
from envied.core.service import Service
from envied.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
from envied.core.tracks import Chapter, Chapters, Subtitle, Track, Tracks
from envied.core.tracks import Audio, Chapter, Chapters, Subtitle, Track, Tracks
class TUBI(Service):
@@ -23,12 +26,17 @@ class TUBI(Service):
Service code for TubiTV streaming service (https://tubitv.com/)
\b
Version: 1.0.3
Version: 1.0.5
Author: stabbedbybrick
Authorization: None
Authorization: Cookies (Optional)
Geofence: Locked to whatever region the user is in (API only)
Robustness:
Widevine:
L3: 720p, AAC2.0
Widevine:
L3: 1080p, AAC2.0
PlayReady:
SL2000: 1080p, AAC2.0
Clear:
1080p, AAC2.0
\b
Tips:
@@ -36,11 +44,13 @@ class TUBI(Service):
/series/300001423/gotham
/tv-shows/200024793/s01-e01-pilot
/movies/589279/the-outsiders
- Use '-v H.265' to request HEVC tracks.
\b
Notes:
- Due to the structure of the DASH manifest and requests downloader failing to output progress,
aria2c is used as the downloader no matter what downloader is specified in the config.
- Authentication is currently not required, but cookies are used if provided.
- If 1080p exists, it's currently only available as H.265.
- Unshackle fails to mux properly when n_m3u8dl_re is used, so aria2c is forced as downloader.
- Search is currently disabled.
"""
@@ -57,6 +67,19 @@ class TUBI(Service):
self.title = title
super().__init__(ctx)
cdm = ctx.obj.cdm
self.drm_system = "playready" if isinstance(cdm, PlayReadyCdm) else "widevine"
vcodec = ctx.parent.params.get("vcodec")
self.vcodec = "H264" if vcodec is None else "H265"
def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential)
self.auth_token = None
if cookies is not None:
self.auth_token = next((cookie.value for cookie in cookies if cookie.name == "at"), None)
self.session.headers.update({"Authorization": f"Bearer {self.auth_token}"})
# Disable search for now
# def search(self) -> Generator[SearchResult, None, None]:
# params = {
@@ -99,12 +122,18 @@ class TUBI(Service):
raise ValueError("Could not parse ID from title - is the URL correct?")
params = {
"platform": "android",
"content_id": content_id,
"app_id": "tubitv",
"platform": "web", # web, android, androidtv
"device_id": str(uuid.uuid4()),
"content_id": content_id,
"limit_resolutions[]": [
"h264_1080p",
"h265_1080p",
],
"video_resources[]": [
"dash_widevine_nonclearlead",
"dash_playready_psshv0",
"dash",
"dash_widevine",
],
}
@@ -175,15 +204,34 @@ class TUBI(Service):
)
def get_tracks(self, title: Title_T) -> Tracks:
if not title.data.get("video_resources"):
if not (resources := title.data.get("video_resources")):
self.log.error(" - Failed to obtain video resources. Check geography settings.")
self.log.info(f"Title is available in: {title.data.get('country')}")
sys.exit(1)
self.manifest = title.data["video_resources"][0]["manifest"]["url"]
self.license = title.data["video_resources"][0].get("license_server", {}).get("url")
codecs = [x.get("codec") for x in resources]
if not any(self.vcodec in x for x in codecs):
raise ValueError(f"Could not find a {self.vcodec} video resource for this title")
tracks = DASH.from_url(url=self.manifest, session=self.session).to_tracks(language=title.language)
resource = next((
x for x in resources
if self.drm_system in x.get("type", "") and self.vcodec in x.get("codec", "")
), None) or next((
x for x in resources
if self.drm_system not in x.get("type", "") and
"dash" in x.get("type", "") and
self.vcodec in x.get("codec", "")
), None)
if not resource:
raise ValueError("Could not find a video resource for this title")
manifest = resource.get("manifest", {}).get("url")
if not manifest:
raise ValueError("Could not find a manifest for this title")
title.data["license_url"] = resource.get("license_server", {}).get("url")
tracks = DASH.from_url(url=manifest, session=self.session).to_tracks(language=title.language)
for track in tracks:
rep_base = track.data["dash"]["representation"].find("BaseURL")
if rep_base is not None:
@@ -193,10 +241,10 @@ class TUBI(Service):
track.descriptor = Track.Descriptor.URL
track.downloader = aria2c
for track in tracks.audio:
role = track.data["dash"]["adaptation_set"].find("Role")
if role is not None and role.get("value") in ["description", "alternative", "alternate"]:
track.descriptive = True
if isinstance(track, Audio):
role = track.data["dash"]["adaptation_set"].find("Role")
if role is not None and role.get("value") in ["description", "alternative", "alternate"]:
track.descriptive = True
if title.data.get("subtitles"):
tracks.add(
@@ -233,12 +281,22 @@ class TUBI(Service):
def get_widevine_service_certificate(self, **_: Any) -> str:
return None
def get_widevine_license(self, challenge: bytes, **_: Any) -> bytes:
if not self.license:
def get_widevine_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
if not (license_url := title.data.get("license_url")):
return None
r = self.session.post(url=self.license, data=challenge)
r = self.session.post(url=license_url, data=challenge)
if r.status_code != 200:
raise ConnectionError(r.text)
return r.content
def get_playready_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
if not (license_url := title.data.get("license_url")):
return None
r = self.session.post(url=license_url, data=challenge)
if r.status_code != 200:
raise ConnectionError(r.text)
return r.content
packages/envied/src/envied/services/TUBI/config.yaml
+1 -1
View File
@@ -1,5 +1,5 @@
endpoints:
content: https://uapi.adrise.tv/cms/content
content: https://uapi.adrise.tv/cms/content # https://content-cdn.production-public.tubi.io/api/v2/content
search: https://search.production-public.tubi.io/api/v1/search