thegame_1980/unshackle-services
1
0
Fork 0
mirror of https://github.com/n0stal6ic/unshackle-services.git synced 2026-07-15 17:40:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

Update __init__.py

Browse Source
This commit is contained in:
n0stal6ic
2026-03-12 14:38:09 -05:00
committed by GitHub
parent 56cd071191
commit 468de5a244
1 changed files with 118 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HULU/__init__.py
+99 -40
View File
@@ -1,8 +1,9 @@
from __future__ import annotations
from http.cookiejar import CookieJar
from typing import Any
import re
import base64
import hashlib
from http.cookiejar import CookieJar
from typing import Any, Optional, Union
import click
import requests
from click import Context
@@ -20,6 +21,7 @@ class HULU(Service):
\b
Authorization: Cookies
Security: UHD@L3, UHD@SL3000
"""
ALIASES = ["HULU"]
@@ -37,27 +39,29 @@ class HULU(Service):
@staticmethod
@click.command(name="HULU", short_help="hulu.com")
@click.argument("title", type=str)
@click.option("-m", "--movie", is_flag=True, default=False, help="Title is a Movie.")
@click.option("-mt", "--mpd-type", type=click.Choice(["new", "old"], case_sensitive=False), default="new", help="Which MPD type to use.")
@click.option(
"-mt", "--mpd-type",
type=click.Choice(["new", "old"], case_sensitive=False),
default="new",
help="Which MPD type to use.",
)
@click.pass_context
def cli(ctx, **kwargs):
return HULU(ctx, **kwargs)
def __init__(self, ctx: Context, title: str, movie: bool, mpd_type: str):
def __init__(self, ctx: Context, title: str, mpd_type: str):
self.title = title
self.movie = movie
self.mpd_type = mpd_type
self.vcodec = ctx.parent.params.get("vcodec")
self.range = ctx.parent.params.get("range_")
self.acodec = ctx.parent.params["acodec"]
self.cdm = ctx.obj.cdm
self.license_url_widevine = None
self.license_url_playready = None
self.license_url_widevine: str | None = None
self.license_url_playready: str | None = None
super().__init__(ctx)
def _fatal(self, msg: str) -> SystemExit:
def _fatal(self, msg: str) -> None:
self.log.error(msg)
return SystemExit(1)
raise SystemExit(1)
@staticmethod
def _safe_json(response) -> dict:
@@ -66,14 +70,36 @@ class HULU(Service):
except Exception:
return {}
def get_titles(self):
if self.movie:
@staticmethod
def _strip_duplicate_representations(mpd: str) -> str:
seen: set[str] = set()
def _dedupe(m: re.Match) -> str:
rid = m.group(1)
if rid in seen:
return ""
seen.add(rid)
return m.group(0)
return re.sub(
r'<Representation[^>]*\bid="([^"]+)"[^>]*>.*?</Representation>',
_dedupe,
mpd,
flags=re.DOTALL,
)
def get_titles(self, *, _force_movie: bool = False):
if _force_movie:
return self._get_movie()
return self._get_series()
def _get_movie(self):
resp = self.session.get(self.config["endpoints"]["movie"].format(id=self.title))
try:
resp.raise_for_status()
except requests.HTTPError as e:
info = self._safe_json(e.response)
raise self._fatal(f" - Failed to get movie {self.title}: {info.get('message', e)}")
self._fatal(f" - Failed to get movie {self.title}: {info.get('message', e)}")
title_data = resp.json()["details"]["vod_items"]["focus"]["entity"]
return Movies([Movie(
@@ -85,6 +111,7 @@ class HULU(Service):
data=title_data,
)])
def _get_series(self):
resp = self.session.get(self.config["endpoints"]["series"].format(id=self.title))
try:
resp.raise_for_status()
@@ -93,14 +120,13 @@ class HULU(Service):
message = info.get("message", "")
if e.response.status_code == 400 and "entity type" in message.lower():
self.log.info(" - Detected movie UUID, retrying on movie endpoint.")
self.movie = True
return self.get_titles()
raise self._fatal(f" - Failed to get series {self.title}: {message} [{info.get('code')}]")
return self.get_titles(_force_movie=True)
self._fatal(f" - Failed to get series {self.title}: {message} [{info.get('code')}]")
res = resp.json()
season_data = next((x for x in res.get("components", []) if x.get("name") == "Episodes"), None)
if not season_data:
raise self._fatal(" - Failed to get episodes.")
self._fatal(" - Failed to get episodes.")
series = Series()
for season in season_data.get("items", []):
@@ -112,7 +138,7 @@ class HULU(Service):
season_resp.raise_for_status()
except requests.HTTPError as e:
info = self._safe_json(e.response)
raise self._fatal(f" - Failed to get season {season_id_part}: {info.get('message', e)}")
self._fatal(f" - Failed to get season {season_id_part}: {info.get('message', e)}")
for episode in season_resp.json().get("items", []):
try:
@@ -136,18 +162,28 @@ class HULU(Service):
return series
def get_tracks(self, title):
codec = "H265" if self.vcodec == Video.Codec.HEVC else "H264"
if self.vcodec == Video.Codec.HEVC:
codec = "H265"
elif self.vcodec == Video.Codec.AVC:
codec = "H264"
else:
self.log.warning(f" - Unrecognised vcodec '{self.vcodec}', defaulting to H264.")
codec = "H264"
eab_id = (title.data.get("bundle") or {}).get("eab_id")
if not eab_id:
raise self._fatal(f" - Could not find eab_id in {title}")
self._fatal(f" - Could not find eab_id in title data for '{title}'.")
device_cfg = self.config["device_ids"]
deejay_id = device_cfg["new"] if self.mpd_type == "new" else device_cfg["old"]
version = 1 if self.mpd_type == "new" else 9999999
try:
resp = self.session.post(
url=self.config["endpoints"]["manifest"],
json={
"deejay_device_id": 210 if self.mpd_type == "new" else 166,
"version": 1 if self.mpd_type == "new" else 9999999,
"deejay_device_id": deejay_id,
"version": version,
"all_cdn": False,
"content_eab_id": eab_id,
"region": "US",
@@ -162,17 +198,21 @@ class HULU(Service):
"codecs": {
"values": [x for x in self.config["codecs"]["video"] if x["type"] == codec],
"selection_mode": self.config["codecs"]["video_selection"],
}
},
},
"audio": {
"codecs": {
"values": self.config["codecs"]["audio"],
"selection_mode": self.config["codecs"]["audio_selection"],
}
},
},
"drm": {
"multi_key": True,
"values": self.config["drm"]["schemas_pr"] if isinstance(self.cdm, PlayReadyCdm) else self.config["drm"]["schemas_wv"],
"values": (
self.config["drm"]["schemas_pr"]
if isinstance(self.cdm, PlayReadyCdm)
else self.config["drm"]["schemas_wv"]
),
"selection_mode": self.config["drm"]["selection_mode"],
"hdcp": self.config["drm"]["hdcp"],
},
@@ -193,20 +233,20 @@ class HULU(Service):
"https": True,
}],
"selection_mode": "ONE",
}
}
}
},
},
},
)
resp.raise_for_status()
playlist = resp.json()
except requests.HTTPError as e:
info = self._safe_json(e.response)
raise self._fatal(f" - Failed to fetch manifest: {info.get('message', e)} ({info.get('code')})")
self._fatal(f" - Failed to fetch manifest: {info.get('message', e)} ({info.get('code')})")
except ValueError as e:
raise self._fatal(f" - Failed to decode manifest JSON: {e}")
self._fatal(f" - Failed to decode manifest JSON: {e}")
if "stream_url" not in playlist:
raise self._fatal(f" - Manifest response missing 'stream_url'. Keys: {list(playlist.keys())}")
self._fatal(f" - Manifest response missing 'stream_url'. Keys: {list(playlist.keys())}")
self.license_url_widevine = playlist.get("wv_server")
self.license_url_playready = playlist.get("dash_pr_server")
@@ -214,7 +254,10 @@ class HULU(Service):
manifest = playlist["stream_url"]
self.log.info(f"DASH: {manifest}")
tracks = DASH.from_url(manifest, self.session).to_tracks(title.language)
mpd_resp = self.session.get(manifest)
mpd_resp.raise_for_status()
mpd_text = self._strip_duplicate_representations(mpd_resp.text)
tracks = DASH.from_text(mpd_text, manifest).to_tracks(title.language)
if self.acodec:
mapped = self.AUDIO_CODEC_MAP.get(self.acodec)
@@ -222,8 +265,8 @@ class HULU(Service):
tracks.audio = [x for x in tracks.audio if (x.codec or "").startswith(mapped)]
for track in tracks.audio:
if track.bitrate > 768000:
track.bitrate = 768000
if track.bitrate > 768_000:
track.bitrate = 768_000
if track.channels == 6.0:
track.channels = 5.1
@@ -239,16 +282,26 @@ class HULU(Service):
return tracks
def get_chapters(self, title: Union[Movie, Episode]) -> list[Chapter]:
def get_chapters(self, title: Movie | Episode) -> list[Chapter]:
return []
def get_widevine_license(self, challenge, track, **_):
def get_widevine_license(self, challenge, track, **_: Any):
try:
resp = self.session.post(
url=self.license_url_widevine,
data=challenge,
headers={"Content-Type": "application/octet-stream"},
)
self.log.debug(f"License HTTP {resp.status_code}, Content-Type: {resp.headers.get('Content-Type')}, Body[:80]: {resp.content[:80]}")
resp.raise_for_status()
except requests.HTTPError as e:
self._fatal(f" - Widevine license request failed: {e}")
self.log.debug(
f"License HTTP {resp.status_code}, "
f"Content-Type: {resp.headers.get('Content-Type')}, "
f"Body[:80]: {resp.content[:80]}"
)
ct = resp.headers.get("Content-Type", "")
if "json" in ct or resp.content[:1] == b"{":
try:
@@ -258,14 +311,20 @@ class HULU(Service):
return base64.b64decode(data[key])
except Exception:
pass
return resp.content
def get_playready_license(self, challenge, track, **_):
def get_playready_license(self, challenge, track, **_: Any):
try:
resp = self.session.post(url=self.license_url_playready, data=challenge)
resp.raise_for_status()
except requests.HTTPError as e:
self._fatal(f" - PlayReady license request failed: {e}")
self.log.debug(resp.content)
return resp.content
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
def authenticate(self, cookies: CookieJar | None = None, credential: Credential | None = None) -> None:
if cookies:
self.session.cookies.update(cookies)
self.session.headers.update({"User-Agent": self.config["user_agent"]})