y0ELGVJtIj/unshackle-services
1
0
Fork 0
forked from FairTrade/unshackle-services
Code Pull Requests Activity

Added series support for VIKI

Browse Source
This commit is contained in:
FairTrade 2025-11-14 20:29:00 +01:00
parent debb3e24fe
commit 99407a7d7d
2 changed files with 106 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -18,7 +18,6 @@ These services is new and in development. Please feel free to submit pull reques
- MUBI
- Search Functionality
- VIKI
- Series support soon
- CSRF Token is now scraped, would be from a api requests soon

123
VIKI/__init__.py
View File

@ -2,7 +2,6 @@ import base64
import json
import os
import re
import xml.etree.ElementTree as ET
from http.cookiejar import CookieJar
from typing import Optional, Generator
@ -21,13 +20,13 @@ from langcodes import Language
class VIKI(Service):
"""
Service code for Rakuten Viki (viki.com)
Version: 1.3.9
Version: 1.4.0
Authorization: Required cookies (_viki_session, device_id).
Security: FHD @ L3 (Widevine)
Supports:
• Movies only
• Movies and TV Series
"""
TITLE_RE = r"^(?:https?://(?:www\.)?viki\.com)?/(?:movies|tv)/(?P<id>\d+c)-.+$"
@ -136,13 +135,109 @@ class VIKI(Service):
)
])
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.")
def _parse_series(self, data: dict) -> Series:
"""Parse series metadata and fetch episodes."""
series_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.log.info(f"Parsing series: {series_name}")
# Fetch episode list IDs
episodes_url = self.config["endpoints"]["episodes"].format(container_id=self.container_id)
params = {
"app": self.config["params"]["app"],
"token": self.api_access_key,
"direction": "asc",
"with_upcoming": "true",
"sort": "number",
"blocked": "true",
"only_ids": "true"
}
r = self.session.get(episodes_url, params=params)
r.raise_for_status()
episodes_data = r.json()
episode_ids = episodes_data.get("response", [])
self.log.info(f"Found {len(episode_ids)} episodes")
episodes = []
for idx, ep_id in enumerate(episode_ids, 1):
# Fetch individual episode metadata
ep_url = self.config["endpoints"]["episode_meta"].format(video_id=ep_id)
ep_params = {
"app": self.config["params"]["app"],
"token": self.api_access_key,
}
try:
r_ep = self.session.get(ep_url, params=ep_params)
r_ep.raise_for_status()
ep_data = r_ep.json()
ep_number = ep_data.get("number", idx)
ep_title = ep_data.get("titles", {}).get("en", "")
ep_description = ep_data.get("descriptions", {}).get("en", "")
# If no episode title, use generic name
if not ep_title:
ep_title = f"Episode {ep_number}"
# Store the video_id in the data dict
ep_data["video_id"] = ep_id
self.log.debug(f"Episode {ep_number}: {ep_title} ({ep_id})")
episodes.append(
Episode(
id_=ep_id,
service=self.__class__,
title=series_name, # Series title
season=1, # VIKI typically doesn't separate seasons clearly
number=ep_number,
name=ep_title, # Episode title
description=ep_description,
language=Language.get(original_lang_code),
data=ep_data
)
)
except Exception as e:
self.log.warning(f"Failed to fetch episode {ep_id}: {e}")
# Create a basic episode entry even if metadata fetch fails
episodes.append(
Episode(
id_=ep_id,
service=self.__class__,
title=series_name,
season=1,
number=idx,
name=f"Episode {idx}",
description="",
language=Language.get(original_lang_code),
data={"video_id": ep_id} # Store video_id in data
)
)
# Return Series with just the episodes list
return Series(episodes)
def get_tracks(self, title: Title_T) -> Tracks:
# For episodes, get the video_id from the data dict
if isinstance(title, Episode):
self.video_id = title.data.get("video_id")
if not self.video_id:
# Fallback to episode id if video_id not in data
self.video_id = title.data.get("id")
elif not self.video_id:
raise RuntimeError("video_id not set. Call get_titles() first.")
if not self.video_id:
raise ValueError("Could not determine video_id for this title")
self.log.info(f"Getting tracks for video ID: {self.video_id}")
url = self.config["endpoints"]["playback"].format(video_id=self.video_id)
r = self.session.get(url)
r.raise_for_status()
@ -178,13 +273,8 @@ class VIKI(Service):
"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
@ -219,9 +309,8 @@ class VIKI(Service):
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..")
raise ValueError("DRM license URL not available.")
r = self.session.post(
self.drm_license_url,
data=challenge,
@ -236,4 +325,4 @@ class VIKI(Service):
yield
def get_chapters(self, title: Title_T) -> list[Chapter]:
return []
return []