import base64 import json import os import re import xml.etree.ElementTree as ET from http.cookiejar import CookieJar from typing import Optional, Generator import click from unshackle.core.search_result import SearchResult 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 Movie, Movies, Series, Episode, Title_T, Titles_T from unshackle.core.tracks import Chapter, Tracks, Subtitle from unshackle.core.drm import Widevine from langcodes import Language class VIKI(Service): """ Service code for Rakuten Viki (viki.com) Version: 1.3.9 Authorization: Required cookies (_viki_session, device_id). Security: FHD @ L3 (Widevine) Supports: • Movies only """ TITLE_RE = r"^(?:https?://(?:www\.)?viki\.com)?/(?:movies|tv)/(?P\d+c)-.+$" GEOFENCE = () NO_SUBTITLES = False @staticmethod @click.command(name="VIKI", short_help="https://viki.com") @click.argument("title", type=str) @click.pass_context def cli(ctx, **kwargs): return VIKI(ctx, **kwargs) def __init__(self, ctx, title: str): super().__init__(ctx) m = re.match(self.TITLE_RE, title) if not m: self.search_term = title self.title_url = None return self.container_id = m.group("id") self.title_url = title self.video_id: Optional[str] = None self.api_access_key: Optional[str] = None self.drm_license_url: Optional[str] = None self.cdm = ctx.obj.cdm if self.config is None: raise EnvironmentError("Missing service config for VIKI.") def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None: super().authenticate(cookies, credential) if not cookies: raise PermissionError("VIKI requires a cookie file for authentication.") session_cookie = next((c for c in cookies if c.name == "_viki_session"), None) device_cookie = next((c for c in cookies if c.name == "device_id"), None) if not session_cookie or not device_cookie: raise PermissionError("Your cookie file is missing '_viki_session' or 'device_id'.") self.session.headers.update({ "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0", "X-Viki-App-Ver": "14.64.0", "X-Viki-Device-ID": device_cookie.value, "Origin": "https://www.viki.com", "Referer": "https://www.viki.com/", }) self.log.info("VIKI authentication cookies loaded successfully.") def get_titles(self) -> Titles_T: if not self.title_url: raise ValueError("No URL provided to process.") self.log.debug(f"Scraping page for API access key: {self.title_url}") r_page = self.session.get(self.title_url) r_page.raise_for_status() match = re.search(r'"token":"([^"]+)"', r_page.text) if not match: raise RuntimeError("Failed to extract API access key from page source.") self.api_access_key = match.group(1) self.log.debug(f"Extracted API access key: {self.api_access_key[:10]}...") url = self.config["endpoints"]["container"].format(container_id=self.container_id) params = { "app": self.config["params"]["app"], "token": self.api_access_key, } r = self.session.get(url, params=params) r.raise_for_status() data = r.json() content_type = data.get("type") if content_type == "film": return self._parse_movie(data) elif content_type == "series": return self._parse_series(data) else: self.log.error(f"Unknown content type '{content_type}' found.") return Movies([]) def _parse_movie(self, data: dict) -> Movies: name = data.get("titles", {}).get("en", "Unknown Title") year = int(data["created_at"][:4]) if "created_at" in data else None description = data.get("descriptions", {}).get("en", "") original_lang_code = data.get("origin", {}).get("language", "en") self.video_id = data.get("watch_now", {}).get("id") if not self.video_id: raise ValueError(f"Could not find a playable video ID for container {self.container_id}.") return Movies([ Movie( id_=self.container_id, service=self.__class__, name=name, year=year, description=description, language=Language.get(original_lang_code), data=data, ) ]) def get_tracks(self, title: Title_T) -> Tracks: if not self.video_id: if isinstance(title, Episode): self.video_id = title.id_ else: raise RuntimeError("video_id not set. Call get_titles() first.") url = self.config["endpoints"]["playback"].format(video_id=self.video_id) r = self.session.get(url) r.raise_for_status() data = r.json() # Get the DRM-protected manifest from queue manifest_url = None for item in data.get("queue", []): if item.get("type") == "video" and item.get("format") == "mpd": manifest_url = item.get("url") break if not manifest_url: raise ValueError("No DRM-protected manifest URL found in queue") self.log.debug(f"Found DRM-protected manifest URL: {manifest_url}") # Create headers for manifest download manifest_headers = { "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0", "Accept": "*/*", "Accept-Language": "en", "Accept-Encoding": "gzip, deflate, br, zstd", "X-Viki-App-Ver": "14.64.0", "X-Viki-Device-ID": self.session.headers.get("X-Viki-Device-ID", ""), "Origin": "https://www.viki.com", "Referer": "https://www.viki.com/", "Connection": "keep-alive", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "cross-site", "Pragma": "no-cache", "Cache-Control": "no-cache", } # Download the DRM-protected manifest manifest_response = self.session.get(manifest_url, headers=manifest_headers) manifest_response.raise_for_status() # Parse tracks from the DRM-protected manifest tracks = DASH.from_url(manifest_url, session=self.session).to_tracks(language=title.language) # Subtitles title_language = title.language.language subtitles = [] for sub in data.get("subtitles", []): sub_url = sub.get("src") lang_code = sub.get("srclang") if not sub_url or not lang_code: continue subtitles.append( Subtitle( id_=lang_code, url=sub_url, language=Language.get(lang_code), is_original_lang=lang_code == title_language, codec=Subtitle.Codec.WebVTT, name=sub.get("label", lang_code.upper()).split(" (")[0] ) ) tracks.subtitles = subtitles # Store DRM license URL (only dt3) at service level drm_b64 = data.get("drm") if drm_b64: drm_data = json.loads(base64.b64decode(drm_b64)) self.drm_license_url = drm_data.get("dt3") # Use dt3 as requested else: self.log.warning("No DRM info found, assuming unencrypted stream.") return tracks def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes: if not hasattr(self, 'drm_license_url') or not self.drm_license_url: raise ValueError("DRM license URL not available..") r = self.session.post( self.drm_license_url, data=challenge, headers={"Content-type": "application/octet-stream"} ) r.raise_for_status() return r.content def search(self) -> Generator[SearchResult, None, None]: self.log.warning("Search not yet implemented for VIKI.") return yield def get_chapters(self, title: Title_T) -> list[Chapter]: return []