From a49f6d5a2d24b8d1e95852b13740d054132f3b3e Mon Sep 17 00:00:00 2001 From: n0stal6ic Date: Sat, 11 Apr 2026 12:32:43 -0500 Subject: [PATCH] Add files via upload Add files via upload Update README.md Delete PCOK/__init__.py --- PCOK/__init__.py | 445 +++++++++++++++++++++++++++++++++++++++++++++++ PCOK/config.yaml | 33 ++++ README.md | 1 + 3 files changed, 479 insertions(+) create mode 100644 PCOK/__init__.py create mode 100644 PCOK/config.yaml diff --git a/PCOK/__init__.py b/PCOK/__init__.py new file mode 100644 index 0000000..1d560b7 --- /dev/null +++ b/PCOK/__init__.py @@ -0,0 +1,445 @@ +from __future__ import annotations +import base64 +import hashlib +import hmac +import json +import re +import time +from datetime import datetime +from http.cookiejar import CookieJar +from typing import Optional +from urllib.parse import urlparse +import click +from langcodes import Language +from unshackle.core.constants import AnyTrack +from unshackle.core.credential import Credential +from unshackle.core.manifests import DASH +from unshackle.core.service import Service +from unshackle.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T +from unshackle.core.tracks import Chapters, Tracks, Video + + +class PCOK(Service): + """ + Service code for Peacock TV (https://peacocktv.com) + + Author: n0stal6ic + Authorization: Cookies, Credentials + Geofence: US + """ + + ALIASES = ("PCOK", "peacock", "peacocktv") + GEOFENCE = ("US",) + TITLE_RE = [ + r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/movies/[a-z0-9_./-]+/[a-f0-9-]{36})", + r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/tv/[a-z0-9_./-]+/[a-f0-9-]{36})", + r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/tv/[a-z0-9_./-]+/\d+)", + r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/news/[a-z0-9_./-]+/[a-f0-9-]{36})", + r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/-/[a-z0-9_./-]+/\d+)", + r"(?:https?://(?:www\.)?peacocktv\.com/stream-tv/)?(?P[a-z0-9-]+)$", + ] + + @staticmethod + @click.command(name="PCOK", short_help="https://peacocktv.com") + @click.argument("title", type=str) + @click.option("-m", "--movie", is_flag=True, default=False, help="Title is a movie.") + @click.pass_context + def cli(ctx, **kwargs): + return PCOK(ctx, **kwargs) + + def __init__(self, ctx, title: str, movie: bool): + super().__init__(ctx) + + self.title = title + self.movie = movie + + range_param = ctx.parent.params.get("range_") + self.range = range_param[0].name if range_param else "SDR" + + vcodec_param = ctx.parent.params.get("vcodec") + self.vcodec = vcodec_param if vcodec_param else "H264" + + self.profile_name = ctx.parent.params.get("profile") or "default" + + prof_key = self.config["client"].get("profile", "tv") + profiles = self.config.get("profiles", {}) + if prof_key not in profiles: + raise ValueError(f"Unknown device profile {prof_key!r}. Valid: {list(profiles)}") + self.prof = profiles[prof_key] + self.hmac_key: bytes = self.prof["hmac_key"].encode() + + try: + from pyplayready.cdm import Cdm as PlayReadyCdm + self.use_playready: bool = isinstance(ctx.obj.cdm, PlayReadyCdm) + except ImportError: + self.use_playready = False + + self.tokens: Optional[dict] = None + self.license_url: Optional[str] = None + + def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None: + super().authenticate(cookies, credential) + self.session.headers.update({"Origin": "https://www.peacocktv.com"}) + + if credential and credential.username and credential.password: + self.log.info("Authenticating with email/password credentials.") + self._login(credential.username, credential.password) + elif not cookies: + raise EnvironmentError("Peacock requires cookies or credential.") + + self.log.info("Fetching authorization tokens.") + self.tokens = self._get_tokens() + self.log.info("Verifying tokens.") + if not self._verify_tokens(): + raise EnvironmentError("Token verification failed.") + + def _login(self, email: str, password: str) -> None: + r = self.session.post( + url=self.config["endpoints"]["login"], + data={"userIdentifier": email, "password": password}, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "X-SkyOTT-Proposition": self.config["client"]["proposition"], + "X-SkyOTT-Provider": self.config["client"]["provider"], + "X-SkyOTT-Territory": self.config["client"]["territory"], + }, + ) + if r.status_code not in (200, 201): + try: + code = ( + r.json() + .get("properties", {}) + .get("errors", {}) + .get("categoryErrors", [{}])[0] + .get("code", "unknown") + ) + raise EnvironmentError(f"Login failed: {code}") + except (ValueError, KeyError, IndexError): + raise EnvironmentError(f"Login failed with HTTP {r.status_code}.") + + def _sky_headers(self, extra: Optional[dict] = None) -> dict: + h = { + "X-SkyOTT-Device": self.prof["device"], + "X-SkyOTT-Platform": self.prof["platform"], + "X-SkyOTT-Proposition": self.config["client"]["proposition"], + "X-SkyOTT-Provider": self.config["client"]["provider"], + "X-SkyOTT-Territory": self.config["client"]["territory"], + } + if extra: + h.update(extra) + return h + + def _md5_headers(self, headers: dict) -> str: + lines = sorted( + f"{k.lower()}: {v}" + for k, v in headers.items() + if k.lower().startswith("x-skyott-") + ) + text = "\n".join(lines) + ("\n" if lines else "") + return hashlib.md5(text.encode()).hexdigest() + + @staticmethod + def _md5_body(body: str | bytes) -> str: + if isinstance(body, str): + body = body.encode() + return hashlib.md5(body).hexdigest() + + def _sign(self, method: str, path: str, headers: dict, body: str | bytes = b"") -> str: + ts = str(int(time.time())) + sdk = self.prof["client_sdk"] + msg = ( + "\n".join([ + method.upper(), + path, + "", + sdk, + "1.0", + self._md5_headers(headers), + ts, + self._md5_body(body), + ]) + + "\n" + ) + digest = hmac.new(self.hmac_key, msg.encode(), hashlib.sha1).digest() + sig = base64.b64encode(digest).decode() + return f'SkyOTT client="{sdk}",signature="{sig}",timestamp="{ts}",version="1.0"' + + def _get_tokens(self) -> dict: + prof_key = self.config["client"].get("profile", "tv") + cache_key = f"tokens_{self.profile_name}_{prof_key}" + cache = self.cache.get(cache_key) + + if cache and cache.data: + expiry = cache.data.get("tokenExpiryTime") + if expiry: + try: + if datetime.strptime(expiry, "%Y-%m-%dT%H:%M:%S.%fZ") > datetime.utcnow(): + self.log.debug("Using cached tokens.") + return cache.data + except ValueError: + pass + + sky_h = self._sky_headers() + + persona_id: Optional[str] = None + try: + r = self.session.get( + url=self.config["endpoints"]["personas"], + headers={ + **sky_h, + "Accept": "application/vnd.persona.v1+json", + "X-SkyOTT-TokenType": self.config["client"]["auth_scheme"], + }, + ) + if r.ok: + personas = r.json().get("personas", []) + if personas: + persona_id = personas[0]["personaId"] + except Exception as e: + self.log.debug(f"Persona fetch skipped: {e}") + + auth_block: dict = { + "authScheme": self.config["client"]["auth_scheme"], + "provider": self.config["client"]["provider"], + "providerTerritory": self.config["client"]["territory"], + "proposition": self.config["client"]["proposition"], + } + if persona_id: + auth_block["personaId"] = persona_id + + body = json.dumps( + { + "auth": auth_block, + "device": { + "type": self.prof["device"], + "platform": self.prof["platform"], + "id": self.config["client"].get("device_id", "PC"), + "drmDeviceId": self.config["client"].get("drm_device_id", "UNKNOWN"), + }, + }, + separators=(",", ":"), + ) + + r = self.session.post( + url=self.config["endpoints"]["tokens"], + data=body, + headers={ + **sky_h, + "Accept": "application/vnd.tokens.v1+json", + "Content-Type": "application/vnd.tokens.v1+json", + "X-Sky-Signature": self._sign("POST", "/auth/tokens", sky_h, body), + }, + ) + r.raise_for_status() + tokens = r.json() + + if "description" in tokens and "userToken" not in tokens: + raise EnvironmentError(f"Token fetch failed: {tokens['description']}") + + if cache: + cache.set(data=tokens) + + tokens["_fresh"] = True + return tokens + + def _verify_tokens(self) -> bool: + if self.tokens.pop("_fresh", False): + return True + sky_h = self._sky_headers({"X-SkyOTT-UserToken": self.tokens["userToken"]}) + try: + r = self.session.get( + url=self.config["endpoints"]["me"], + headers={ + **sky_h, + "Accept": "application/vnd.userinfo.v2+json", + "X-Sky-Signature": self._sign("GET", "/auth/users/me", sky_h), + }, + ) + return r.ok + except Exception: + return False + + def get_titles(self) -> Titles_T: + title_id = self.title + for pattern in self.TITLE_RE: + m = re.search(pattern, self.title) + if m: + title_id = m.group("id") + break + + if "/" not in title_id: + r = self.session.get(f"https://www.peacocktv.com/stream-tv/{title_id}") + m = re.search(r"/watch/asset(/[^'\"?#\s]+)", r.text) + if m: + title_id = m.group(1) + else: + raise ValueError(f"Could not resolve slug: {title_id!r}") + + if not title_id.startswith("/"): + title_id = f"/{title_id}" + + if title_id.startswith("/movies/"): + self.movie = True + + sky_h = self._sky_headers() + res = self.session.get( + url=self.config["endpoints"]["node"], + params={"slug": title_id, "represent": "(items(items))"}, + headers={ + **sky_h, + "Accept": "*", + "Referer": f"https://www.peacocktv.com/watch/asset{title_id}", + "X-SkyOTT-Language": "en", + }, + ).json() + + if self.movie: + return Movies([ + Movie( + id_=title_id, + service=self.__class__, + name=res["attributes"]["title"], + year=res["attributes"].get("year"), + data=res, + ) + ]) + + episodes = [ + ep + for season in res.get("relationships", {}).get("items", {}).get("data", []) + for ep in season.get("relationships", {}).get("items", {}).get("data", []) + ] + + return Series([ + Episode( + id_=title_id, + service=self.__class__, + title=res["attributes"]["title"], + season=ep["attributes"].get("seasonNumber"), + number=ep["attributes"].get("episodeNumber"), + name=ep["attributes"].get("title"), + year=ep["attributes"].get("year"), + data=ep, + ) + for ep in episodes + ]) + + def get_tracks(self, title: Title_T) -> Tracks: + attrs = title.data["attributes"] + formats = attrs.get("formats", {}) + + want_uhd = self.vcodec == "H265" + if want_uhd and "UHD" in formats: + content_id = formats["UHD"]["contentId"] + elif "HD" in formats: + content_id = formats["HD"]["contentId"] + else: + content_id = next(iter(formats.values()), {}).get("contentId", "") + + variant_id = attrs.get("providerVariantId", "") + + if self.range == "HDR10": + colour_spaces = ["HDR10"] + elif self.range == "DV": + colour_spaces = ["DolbyVision"] + else: + colour_spaces = ["SDR"] + + primary_drm = "PLAYREADY" if self.use_playready else "WIDEVINE" + + capabilities = [ + { + "protection": primary_drm, + "container": "ISOBMFF", + "transport": "DASH", + "acodec": "AAC", + "vcodec": self.vcodec, + } + ] + + sky_h = self._sky_headers({"X-SkyOTT-UserToken": self.tokens["userToken"]}) + body = json.dumps( + { + "device": { + "capabilities": capabilities, + "maxVideoFormat": "UHD" if want_uhd else "HD", + "supportedColourSpaces": colour_spaces, + "model": self.prof["platform"], + "hdcpEnabled": "true", + }, + "client": {"thirdParties": ["FREEWHEEL", "YOSPACE"]}, + "contentId": content_id, + "providerVariantId": variant_id, + "parentalControlPin": "null", + "personaParentalControlRating": 9, + }, + separators=(",", ":"), + ) + + r = self.session.post( + url=self.config["endpoints"]["vod"], + data=body, + headers={ + **sky_h, + "Accept": "application/vnd.playvod.v1+json", + "Content-Type": "application/vnd.playvod.v1+json", + "X-Sky-Signature": self._sign("POST", "/video/playouts/vod", sky_h, body), + }, + ) + manifest = r.json() + + if "errorCode" in manifest: + raise ValueError( + f"Playout error: {manifest.get('description', 'unknown')} [{manifest['errorCode']}]" + ) + + self.license_url = manifest["protection"]["licenceAcquisitionUrl"] + + endpoints = manifest["asset"]["endpoints"] + dash_url = next( + (e["url"] for e in endpoints if e.get("cdn", "").upper() == "FASTLY"), + endpoints[0]["url"] if endpoints else None, + ) + if not dash_url: + raise ValueError("No DASH endpoint in playout response.") + + tracks = DASH.from_url(url=dash_url, session=self.session).to_tracks(language=Language.get("en")) + + for video in tracks.videos: + if colour_spaces == ["HDR10"]: + video.range = Video.Range.HDR10 + elif colour_spaces == ["DolbyVision"]: + video.range = Video.Range.DV + else: + video.range = Video.Range.SDR + + for audio in tracks.audio: + if audio.language.territory == "AD": + audio.language.territory = None + + return tracks + + def get_chapters(self, title: Title_T) -> Chapters: + return Chapters() + + def _license_request(self, challenge: bytes) -> bytes: + path = urlparse(self.license_url).path + r = self.session.post( + url=self.license_url, + data=challenge, + headers={ + "X-Sky-Signature": self._sign("POST", path, {}, challenge), + }, + ) + r.raise_for_status() + return r.content + + def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]: + if not self.license_url: + return None + return self._license_request(challenge) + + def get_playready_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]: + if not self.license_url: + return None + return self._license_request(challenge) \ No newline at end of file diff --git a/PCOK/config.yaml b/PCOK/config.yaml new file mode 100644 index 0000000..8556571 --- /dev/null +++ b/PCOK/config.yaml @@ -0,0 +1,33 @@ +profiles: + tv: + platform: ANDROIDTV + device: TV + client_sdk: NBCU-ANDROID-v3 + hmac_key: JuLQgyFz9n89D9pxcN6ZWZXKWfgj2PNBUb32zybj + android: + platform: ANDROID + device: TABLET + client_sdk: NBCU-ANDROID-v3 + hmac_key: JuLQgyFz9n89D9pxcN6ZWZXKWfgj2PNBUb32zybj + web: + platform: PC + device: COMPUTER + client_sdk: NBCU-WEB-v4 + hmac_key: FvT9VtwvhtSZvqnExMsvDDTEvBqR3HdsMcBFtWYV + +client: + territory: US + provider: NBCU + proposition: NBCUOTT + auth_scheme: MESSO + profile: tv + device_id: PC + drm_device_id: UNKNOWN + +endpoints: + login: https://rango.id.peacocktv.com/signin/service/international + personas: https://persona.id.peacocktv.com/persona-store/personas + tokens: https://ovp.peacocktv.com/auth/tokens + me: https://ovp.peacocktv.com/auth/users/me + node: https://atom.peacocktv.com/adapter-calypso/v3/query/node + vod: https://ovp.peacocktv.com/video/playouts/vod \ No newline at end of file diff --git a/README.md b/README.md index 44038f2..ed12129 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ Service scripts for [Unshackle](https://github.com/unshackle-dl/unshackle). | NFBC | [National Film Board
of Canada](https://www.nfb.ca/) | None | None | CA, US | Finished | | PBS | [PBS](https://www.pbs.org/) | None | Cookies | US | Finished | | PBSK | [PBS Kids](https://pbskids.org/) | FHD@L3 | None | US | Finished | +| PCOK | [Peacock](https://www.peacocktv.com/) | FHD@L3/SL3K
UHD@L1/SL150 | Cookies, Credentials | US | Finished | | XUMO | [XUMO Play](https://play.xumo.com/) | None | None | US | Finished | | HULU | [Hulu](https://www.hulu.com) | FHD@L3/SL2K
UHD@L1/SL3K | Cookies | US | Unfinished | | KNPY | [Kanopy](https://www.kanopy.com/) | FHD@L3 | Cookies, Credentials | US, CA, AU, NZ, GB | Unfinished |