Add KIJK/__init__.py

KIJK/__init__.py

@@ -0,0 +1,128 @@
+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