385 lines
14 KiB
Python
385 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import sys
|
|
import time
|
|
from collections.abc import Generator
|
|
from http.cookiejar import CookieJar
|
|
from typing import Any, Optional, Union
|
|
from urllib.parse import urlparse
|
|
|
|
import click
|
|
import requests
|
|
|
|
from devine.core.credential import Credential
|
|
from devine.core.manifests import DASH
|
|
from devine.core.search_result import SearchResult
|
|
from devine.core.service import Service
|
|
from devine.core.titles import Episode, Movie, Movies, Series, Title_T
|
|
from devine.core.tracks import Chapters, Tracks
|
|
|
|
|
|
class NOW(Service):
|
|
"""
|
|
\b
|
|
Service code for Now TV's streaming service (https://nowtv.com)
|
|
Only UK is currently supported
|
|
|
|
\b
|
|
Authorization: Cookies
|
|
Robustness:
|
|
Widevine:
|
|
L1: 2160p, 1080p, DDP5.1
|
|
L3: 720p, AAC2.0
|
|
|
|
\b
|
|
Tips:
|
|
- Input should be the slug of the title, e.g.:
|
|
/house-of-the-dragon/iYEQZ2rcf32XRKvQ5gm2Aq
|
|
/five-nights-at-freddys-2023/A5EK6sKrAaye7uXVJ57V7
|
|
"""
|
|
|
|
ALIASES = ("nowtv",)
|
|
GEOFENCE = ("gb",)
|
|
|
|
@staticmethod
|
|
@click.command(name="NOW", short_help="https://nowtv.com", help=__doc__)
|
|
@click.argument("title", type=str)
|
|
@click.pass_context
|
|
def cli(ctx, **kwargs):
|
|
return NOW(ctx, **kwargs)
|
|
|
|
def __init__(self, ctx, title):
|
|
self.title = title
|
|
super().__init__(ctx)
|
|
|
|
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
|
|
super().authenticate(cookies, credential)
|
|
if not cookies:
|
|
raise EnvironmentError("Service requires Cookies for Authentication.")
|
|
|
|
self.session.cookies.update(cookies)
|
|
self.persona_id = self.persona()
|
|
|
|
def search(self) -> Generator[SearchResult, None, None]:
|
|
headers = {
|
|
"x-skyott-device": self.config["client"]["device"],
|
|
"x-skyott-language": "en",
|
|
"x-skyott-platform": self.config["client"]["platform"],
|
|
"x-skyott-proposition": self.config["client"]["proposition"],
|
|
"x-skyott-provider": self.config["client"]["provider"],
|
|
"x-skyott-territory": self.config["client"]["territory"],
|
|
}
|
|
|
|
params = {
|
|
"term": self.title,
|
|
"limit": "30",
|
|
}
|
|
|
|
r = self.session.get(self.config["endpoints"]["search"], params=params, headers=headers)
|
|
if r.status_code != 200:
|
|
self.log.error(r.text)
|
|
return
|
|
|
|
for result in r.json()["search"]["results"]:
|
|
yield SearchResult(
|
|
id_=result.get("slug"),
|
|
title=result.get("title"),
|
|
description=result.get("description"),
|
|
label=result["channel"].get("name"),
|
|
url="https://www.nowtv.com/gb/watch/home/asset" + result.get("slug"),
|
|
)
|
|
|
|
def get_titles(self) -> Union[Movies, Series]:
|
|
if not self.title.startswith("/"):
|
|
self.title = "/" + self.title
|
|
|
|
res = self.session.get(
|
|
url=self.config["endpoints"]["node"],
|
|
params={"slug": self.title, "represent": "(items(items))"},
|
|
headers={
|
|
"Accept": "*",
|
|
"X-SkyOTT-Device": self.config["client"]["device"],
|
|
"X-SkyOTT-Platform": self.config["client"]["platform"],
|
|
"X-SkyOTT-Proposition": self.config["client"]["proposition"],
|
|
"X-SkyOTT-Provider": self.config["client"]["provider"],
|
|
"X-SkyOTT-Territory": self.config["client"]["territory"],
|
|
},
|
|
).json()
|
|
|
|
if "MOVIES" in res["attributes"].get("classification", ""):
|
|
return Movies(
|
|
[
|
|
Movie(
|
|
id_=self.title,
|
|
name=res["attributes"]["title"],
|
|
year=res["attributes"]["year"],
|
|
service=self.__class__,
|
|
language="en-GB",
|
|
data=res,
|
|
)
|
|
]
|
|
)
|
|
else:
|
|
titles = [
|
|
episode
|
|
for season in res["relationships"]["items"]["data"]
|
|
for episode in season["relationships"]["items"]["data"]
|
|
]
|
|
return Series(
|
|
[
|
|
Episode(
|
|
id_=self.title,
|
|
title=res["attributes"]["title"],
|
|
year=episode["attributes"].get("year"),
|
|
season=episode["attributes"].get("seasonNumber", 0),
|
|
number=episode["attributes"].get("episodeNumber", 0),
|
|
name=episode["attributes"].get("title"),
|
|
service=self.__class__,
|
|
language="en-GB",
|
|
data=episode,
|
|
)
|
|
for episode in titles
|
|
]
|
|
)
|
|
|
|
def get_tracks(self, title: Union[Movies, Series]) -> Tracks:
|
|
variant_id = title.data["attributes"]["providerVariantId"]
|
|
url = self.config["endpoints"]["vod"]
|
|
|
|
headers = {
|
|
"accept": "application/vnd.playvod.v1+json",
|
|
"content-type": "application/vnd.playvod.v1+json",
|
|
"x-skyott-activeterritory": self.config["client"]["territory"],
|
|
"x-skyott-device": self.config["client"]["device"],
|
|
"x-skyott-platform": self.config["client"]["platform"],
|
|
"x-skyott-proposition": self.config["client"]["proposition"],
|
|
"x-skyott-provider": self.config["client"]["provider"],
|
|
"x-skyott-territory": self.config["client"]["territory"],
|
|
"x-skyott-usertoken": self.get_token(),
|
|
}
|
|
|
|
data = {
|
|
"device": {
|
|
"capabilities": [
|
|
# H265 EAC3
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H265",
|
|
"acodec": "EAC3",
|
|
"container": "TS",
|
|
},
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H265",
|
|
"acodec": "EAC3",
|
|
"container": "ISOBMFF",
|
|
},
|
|
{
|
|
"container": "MP4",
|
|
"vcodec": "H265",
|
|
"acodec": "EAC3",
|
|
"protection": "WIDEVINE",
|
|
"transport": "DASH",
|
|
},
|
|
# H264 EAC3
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H264",
|
|
"acodec": "EAC3",
|
|
"container": "TS",
|
|
},
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H264",
|
|
"acodec": "EAC3",
|
|
"container": "ISOBMFF",
|
|
},
|
|
{
|
|
"container": "MP4",
|
|
"vcodec": "H264",
|
|
"acodec": "EAC3",
|
|
"protection": "WIDEVINE",
|
|
"transport": "DASH",
|
|
},
|
|
# H265 AAC
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H265",
|
|
"acodec": "AAC",
|
|
"container": "TS",
|
|
},
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H265",
|
|
"acodec": "AAC",
|
|
"container": "ISOBMFF",
|
|
},
|
|
{
|
|
"container": "MP4",
|
|
"vcodec": "H265",
|
|
"acodec": "AAC",
|
|
"protection": "WIDEVINE",
|
|
"transport": "DASH",
|
|
},
|
|
# H264 AAC
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H264",
|
|
"acodec": "AAC",
|
|
"container": "TS",
|
|
},
|
|
{
|
|
"transport": "DASH",
|
|
"protection": "WIDEVINE",
|
|
"vcodec": "H264",
|
|
"acodec": "AAC",
|
|
"container": "ISOBMFF",
|
|
},
|
|
{
|
|
"container": "MP4",
|
|
"vcodec": "H264",
|
|
"acodec": "AAC",
|
|
"protection": "WIDEVINE",
|
|
"transport": "DASH",
|
|
},
|
|
],
|
|
"model": self.config["client"]["model"],
|
|
"maxVideoFormat": "SD", # "HD", "UHD"
|
|
"hdcpEnabled": "false",
|
|
"supportedColourSpaces": ["DV", "HDR10", "SDR"],
|
|
},
|
|
"providerVariantId": variant_id,
|
|
"parentalControlPin": "null",
|
|
}
|
|
|
|
data = json.dumps(data)
|
|
headers["x-sky-signature"] = self.calculate_signature("POST", url, headers, data)
|
|
|
|
response = self.session.post(url, headers=headers, data=data).json()
|
|
if response.get("errorCode"):
|
|
self.log.error(response.get("description"))
|
|
sys.exit(1)
|
|
|
|
manifest = response["asset"]["endpoints"][0]["url"]
|
|
self.license = response["protection"]["licenceAcquisitionUrl"]
|
|
locale = response["asset"].get("audioTracks", [])[0].get("locale", "en-GB")
|
|
|
|
tracks = DASH.from_url(url=manifest, session=self.session).to_tracks(language=locale)
|
|
|
|
return tracks
|
|
|
|
def get_chapters(self, title: Title_T) -> Chapters:
|
|
return Chapters()
|
|
|
|
def get_widevine_service_certificate(self, **_: Any) -> str:
|
|
return None # WidevineCdm.common_privacy_cert
|
|
|
|
def get_widevine_license(self, challenge: bytes, **_: Any) -> bytes:
|
|
r = requests.post(url=self.license, data=challenge)
|
|
if r.status_code != 200:
|
|
self.log.error(r.text)
|
|
sys.exit(1)
|
|
return r.content
|
|
|
|
# service specific functions
|
|
|
|
@staticmethod
|
|
def calculate_sky_header(headers: dict) -> str:
|
|
text_headers = ""
|
|
for key in sorted(headers.keys()):
|
|
if key.lower().startswith("x-skyott"):
|
|
text_headers += key + ": " + headers[key] + "\n"
|
|
return hashlib.md5(text_headers.encode()).hexdigest()
|
|
|
|
def calculate_signature(self, method: str, url: str, headers: dict, payload: str) -> str:
|
|
to_hash = (
|
|
"{method}\n{path}\n{response_code}\n{app_id}\n{version}\n{headers_md5}\n" "{timestamp}\n{payload_md5}\n"
|
|
).format(
|
|
method=method,
|
|
path=urlparse(url).path if url.startswith("http") else url,
|
|
response_code="",
|
|
app_id=self.config["client"]["client_sdk"],
|
|
version="1.0",
|
|
headers_md5=self.calculate_sky_header(headers),
|
|
timestamp=int(time.time()),
|
|
payload_md5=hashlib.md5(payload.encode()).hexdigest(),
|
|
)
|
|
|
|
signature_key = bytes(self.config["security"]["signature_hmac_key_v4"], "utf-8")
|
|
hashed = hmac.new(signature_key, to_hash.encode("utf8"), hashlib.sha1).digest()
|
|
signature_hmac = base64.b64encode(hashed).decode("utf8")
|
|
|
|
return self.config["security"]["signature_format"].format(
|
|
client=self.config["client"]["client_sdk"], signature=signature_hmac, timestamp=int(time.time())
|
|
)
|
|
|
|
def get_token(self) -> str:
|
|
url = self.config["endpoints"]["tokens"]
|
|
|
|
headers = {
|
|
"accept": "application/vnd.tokens.v1+json",
|
|
"content-type": "application/vnd.tokens.v1+json",
|
|
"x-skyott-device": self.config["client"]["device"],
|
|
"x-skyott-platform": self.config["client"]["platform"],
|
|
"x-skyott-proposition": self.config["client"]["proposition"],
|
|
"x-skyott-provider": self.config["client"]["provider"],
|
|
"x-skyott-territory": self.config["client"]["territory"],
|
|
}
|
|
|
|
data = {
|
|
"auth": {
|
|
"authScheme": self.config["client"]["auth_scheme"],
|
|
"authToken": self.session.cookies.get("skyCEsidismesso01"),
|
|
"authIssuer": self.config["client"]["auth_issuer"],
|
|
"personaId": self.persona_id,
|
|
"provider": self.config["client"]["provider"],
|
|
"providerTerritory": self.config["client"]["territory"],
|
|
"proposition": self.config["client"]["proposition"],
|
|
},
|
|
"device": {
|
|
"type": self.config["client"]["device"],
|
|
"platform": self.config["client"]["platform"],
|
|
"id": self.config["client"]["id"],
|
|
"drmDeviceId": self.config["client"]["drm_device_id"],
|
|
},
|
|
}
|
|
|
|
data = json.dumps(data)
|
|
headers["Content-MD5"] = hashlib.md5(data.encode("utf-8")).hexdigest()
|
|
|
|
response = self.session.post(url, headers=headers, data=data).json()
|
|
if response.get("message"):
|
|
self.log.error(f"{response['message']}")
|
|
sys.exit(1)
|
|
|
|
return response["userToken"]
|
|
|
|
def persona(self):
|
|
headers = {
|
|
"accept": "application/vnd.persona.v1+json",
|
|
"x-skyid-token": self.session.cookies.get("skyCEsidismesso01"),
|
|
"x-skyott-device": self.config["client"]["device"],
|
|
"x-skyott-platform": self.config["client"]["platform"],
|
|
"x-skyott-proposition": "NOWTV",
|
|
"x-skyott-provider": "NOWTV",
|
|
"x-skyott-territory": self.config["client"]["territory"],
|
|
"x-skyott-tokentype": "SSO",
|
|
}
|
|
response = self.session.get(self.config["endpoints"]["personas"], headers=headers).json()
|
|
if response.get("message"):
|
|
self.log.error(f"{response['message']} - Cookies may have expired")
|
|
sys.exit(1)
|
|
|
|
return response["personas"][0]["personaId"]
|