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) www.nostalgic.cc 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/sports/[a-z0-9_./-]+/[a-f0-9-]{36})", r"(?:https?://(?:www\.)?peacocktv\.com/watch/asset)?(?P/sports/[a-z0-9_./-]+/\d+)", 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.pass_context def cli(ctx, **kwargs): return PCOK(ctx, **kwargs) def __init__(self, ctx, title: str): super().__init__(ctx) self.title = title self.movie = False 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[0] if vcodec_param else "h264" self.profile_name = ctx.parent.params.get("profile") or "default" self.prof_key = self.config["client"].get("profile", "tv") profiles = self.config.get("profiles", {}) if self.prof_key not in profiles: raise ValueError(f"Unknown device profile {self.prof_key!r}. Valid: {list(profiles)}") self.prof = profiles[self.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 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") ) except (ValueError, KeyError, IndexError): code = f"HTTP {r.status_code}" raise EnvironmentError(f"Login failed: {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: cache_key = f"tokens_{self.profile_name}_{self.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, description=res["attributes"].get("synopsis"), ) ]) 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, description=ep["attributes"].get("synopsis"), ) for ep in episodes ]) def get_tracks(self, title: Title_T) -> Tracks: attrs = title.data["attributes"] formats = attrs.get("formats", {}) want_uhd = self.vcodec.lower() in ("hevc", "h.265") 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": "H265" if want_uhd else "H264", } ] sky_h = { "X-SkyOTT-Agent": ".".join([ self.config["client"]["proposition"], self.prof["device"], self.prof["platform"], ]).lower(), "X-SkyOTT-PinOverride": "false", "X-SkyOTT-Provider": self.config["client"]["provider"], "X-SkyOTT-Territory": self.config["client"]["territory"], "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']}]" ) 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.descriptive = True audio.language = Language.make(language=audio.language.language) audio.name = None for track in tracks: track.data["license_url"] = license_url return tracks def get_chapters(self, title: Title_T) -> Chapters: return Chapters() def _license_request(self, challenge: bytes, track: AnyTrack) -> bytes: license_url = track.data["license_url"] path = urlparse(license_url).path r = self.session.post( url=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 track.data.get("license_url"): return None return self._license_request(challenge, track) def get_playready_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[bytes]: if not track.data.get("license_url"): return None return self._license_request(challenge, track)