129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
|
import re
|
||
|
|
from collections.abc import Generator
|
||
|
|
from typing import Optional, Union
|
||
|
|
import urllib.parse
|
||
|
|
import json
|
||
|
|
|
||
|
|
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.search_result import SearchResult
|
||
|
|
from unshackle.core.service import Service
|
||
|
|
from unshackle.core.titles import Movie, Movies, Title_T, Titles_T
|
||
|
|
from unshackle.core.tracks import Tracks, Chapter
|
||
|
|
|
||
|
|
|
||
|
|
class KIJK(Service):
|
||
|
|
"""
|
||
|
|
Service code for kijk.nl
|
||
|
|
Version: 1.0.0
|
||
|
|
|
||
|
|
Authorization: None
|
||
|
|
|
||
|
|
Security: FHD@L3, UHD@L3
|
||
|
|
"""
|
||
|
|
|
||
|
|
TITLE_RE = r"https?://(?:www\.)?kijk\.nl/programmas/[^/]+/([^/?]+)"
|
||
|
|
GEOFENCE = ("NL",)
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
@click.command(name="KIJK", short_help="https://kijk.nl")
|
||
|
|
@click.argument("title", type=str)
|
||
|
|
@click.pass_context
|
||
|
|
def cli(ctx, **kwargs):
|
||
|
|
return KIJK(ctx, **kwargs)
|
||
|
|
|
||
|
|
def __init__(self, ctx, title):
|
||
|
|
super().__init__(ctx)
|
||
|
|
self.title = title
|
||
|
|
if self.config is None:
|
||
|
|
raise Exception("Config is missing!")
|
||
|
|
|
||
|
|
self.session.headers.update({"user-agent": self.config["client"]["default"]["user_agent"]})
|
||
|
|
self.token = None
|
||
|
|
self.license_url = None
|
||
|
|
|
||
|
|
def authenticate(self, cookies=None, credential=None):
|
||
|
|
super().authenticate(cookies, credential)
|
||
|
|
|
||
|
|
self.log.info("Retrieving new token")
|
||
|
|
query = {
|
||
|
|
"query": "query DrmTokenQuery($provider: DrmProvider) {\n drmToken(drmProvider: $provider) {\n expiration\n token\n }\n }",
|
||
|
|
"variables": {
|
||
|
|
"provider": "JWP"
|
||
|
|
}
|
||
|
|
}
|
||
|
|
res = self.session.post(self.config["endpoints"]["graphql"], json=query)
|
||
|
|
res.raise_for_status()
|
||
|
|
self.token = res.json()["data"]["drmToken"]["token"]
|
||
|
|
|
||
|
|
|
||
|
|
def search(self) -> Generator[SearchResult, None, None]:
|
||
|
|
raise NotImplementedError("Search is not supported for this service.")
|
||
|
|
|
||
|
|
def get_titles(self) -> Titles_T:
|
||
|
|
guid_match = re.match(self.TITLE_RE, self.title)
|
||
|
|
if not guid_match:
|
||
|
|
raise ValueError("Invalid KIJK URL. Could not extract GUID.")
|
||
|
|
|
||
|
|
guid = guid_match.group(1)
|
||
|
|
|
||
|
|
query_graphql = "query GetVideoQuery($guid:[String]){programs(guid:$guid){items{guid type metadata availableRegion ...Media ...Tracks ...Sources}}}fragment Media on Program{media{type availableDate availabilityState airedDateTime expirationDate}}fragment Tracks on Program{tracks{file kind label}}fragment Sources on Program{sources{type file drm}}"
|
||
|
|
variables_graphql = json.dumps({"guid": guid})
|
||
|
|
|
||
|
|
url = f"{self.config['endpoints']['graphql']}?query={urllib.parse.quote(query_graphql)}&variables={urllib.parse.quote(variables_graphql)}"
|
||
|
|
|
||
|
|
res = self.session.get(url)
|
||
|
|
res.raise_for_status()
|
||
|
|
|
||
|
|
metadata = res.json()["data"]["programs"]["items"][0]
|
||
|
|
|
||
|
|
return Movies(
|
||
|
|
[
|
||
|
|
Movie(
|
||
|
|
id_=metadata["guid"],
|
||
|
|
service=self.__class__,
|
||
|
|
name=metadata["metadata"]["media_program_name"],
|
||
|
|
description=metadata["metadata"].get("media_description", ""),
|
||
|
|
year=int(metadata["media"][0]["airedDateTime"].split('-')[0]),
|
||
|
|
language=Language.get("nl"), # Hardcoded as it's a Dutch service
|
||
|
|
data=metadata,
|
||
|
|
)
|
||
|
|
]
|
||
|
|
)
|
||
|
|
|
||
|
|
def get_tracks(self, title: Title_T) -> Tracks:
|
||
|
|
dash_link = None
|
||
|
|
for source in title.data["sources"]:
|
||
|
|
if source.get("type") == "dash" and source.get("drm") and "widevine" in source.get("drm"):
|
||
|
|
dash_link = source["file"]
|
||
|
|
self.license_url = source["drm"]["widevine"]["url"]
|
||
|
|
break
|
||
|
|
|
||
|
|
if not dash_link:
|
||
|
|
raise ValueError("Could not find a DASH manifest for this title.")
|
||
|
|
|
||
|
|
self.log.debug(f"Manifest URL: {dash_link}")
|
||
|
|
tracks = DASH.from_url(url=dash_link, session=self.session).to_tracks(language=title.language)
|
||
|
|
|
||
|
|
return tracks
|
||
|
|
|
||
|
|
def get_chapters(self, title: Title_T) -> list[Chapter]:
|
||
|
|
return []
|
||
|
|
|
||
|
|
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
|
||
|
|
if not self.license_url:
|
||
|
|
raise ValueError("Widevine license endpoint not configured")
|
||
|
|
|
||
|
|
headers = {'x-vudrm-token': self.token} if self.token else {}
|
||
|
|
response = self.session.post(
|
||
|
|
url=self.license_url,
|
||
|
|
data=challenge,
|
||
|
|
headers=headers
|
||
|
|
)
|
||
|
|
response.raise_for_status()
|
||
|
|
return response.content
|