mirror of
https://github.com/vinefeeder/TwinVine.git
synced 2026-07-15 18:10:04 +02:00
upstream services
This commit is contained in:
VineFeeder
@@ -18,7 +18,7 @@ tag_imdb_tmdb: true
|
||||
set_terminal_bg: false
|
||||
|
||||
# Custom output templates for filenames
|
||||
# Configure output_template in your unshackle.yaml to control filename format.
|
||||
# Configure output_template in your envied.yaml to control filename format.
|
||||
# If not configured, default scene-style templates are used and a warning is shown.
|
||||
# Available variables: {title}, {year}, {season}, {episode}, {season_episode}, {episode_name},
|
||||
# {quality}, {resolution}, {source}, {audio}, {audio_channels}, {audio_full},
|
||||
@@ -404,7 +404,7 @@ headers:
|
||||
filenames:
|
||||
debug_log: "unshackle_debug_{service}_{time}.jsonl" # JSON Lines debug log file
|
||||
config: "config.yaml"
|
||||
root_config: "unshackle.yaml"
|
||||
root_config: "envied.yaml"
|
||||
chapters: "Chapters_{title}_{random}.txt"
|
||||
subtitle: "Subtitle_{id}_{language}.srt"
|
||||
|
||||
|
||||
@@ -117,6 +117,7 @@ key_vaults:
|
||||
#- type: SQLite
|
||||
# name: Local vault
|
||||
# path: key_store.db
|
||||
|
||||
- type: HTTPAPI
|
||||
name: drmlab
|
||||
host: http://api.drmlab.io/vault/
|
||||
|
||||
@@ -0,0 +1,605 @@
|
||||
import base64
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from http.cookiejar import CookieJar
|
||||
from typing import Any
|
||||
|
||||
import click
|
||||
from envied.core.credential import Credential
|
||||
from envied.core.manifests import DASH
|
||||
from envied.core.service import Service
|
||||
from envied.core.session import session
|
||||
from envied.core.titles import Episode, Movie, Movies, Series
|
||||
from envied.core.tracks import Audio, Chapter, Chapters, Subtitle, Tracks
|
||||
from envied.core.cdm.detect import is_playready_cdm
|
||||
|
||||
|
||||
class CRAV(Service):
|
||||
"""
|
||||
Service code for Bell Media's Crave streaming service (https://crave.ca).
|
||||
|
||||
\b
|
||||
Version: 1.0.0
|
||||
Author: stabbedbybrick
|
||||
Authorization: Credentials
|
||||
Geofence: CA (API and downloads)
|
||||
Robustness:
|
||||
Widevine:
|
||||
L1: 1080p, 2160p
|
||||
L3: 720p
|
||||
PlayReady:
|
||||
SL150: 2160p
|
||||
|
||||
\b
|
||||
Tips:
|
||||
- Input examples:
|
||||
SERIES: https://www.crave.ca/en/series/it-welcome-to-derry-58324
|
||||
EPISODE: https://www.crave.ca/en/play/it-welcome-to-derry/the-pilot-s1e1-3208281
|
||||
MOVIE: https://www.crave.ca/en/movie/superman-2025-58844
|
||||
|
||||
\b
|
||||
Notes:
|
||||
- Subtitles are not always consistent, so both external (VTT) and internal (WVTT) subtitles are added.
|
||||
If you experience incomplete subtitles, try using a newer version of Unshackle (v5.0.0+).
|
||||
- Authentication will look for master profile first, then first profile with adult maturity scope.
|
||||
- Account pins are currently not supported.
|
||||
"""
|
||||
|
||||
ALIASES = ("crave",)
|
||||
GEOFENCE = ("ca",)
|
||||
TITLE_RE = r"^(?:https?://(?:www\.)?crave\.ca(?:/[a-z]{2})?/(?P<type>movie|series|play)/(?:[a-z0-9-]+/)?)?([a-z0-9-]+?)(?:-(?P<id>\d+))?$"
|
||||
|
||||
@staticmethod
|
||||
@click.command(name="CRAV", short_help="https://crave.ca")
|
||||
@click.argument("title", type=str)
|
||||
@click.pass_context
|
||||
def cli(ctx: click.Context, **kwargs: Any) -> "CRAV":
|
||||
return CRAV(ctx, **kwargs)
|
||||
|
||||
def __init__(self, ctx, title):
|
||||
super().__init__(ctx)
|
||||
self.title = title
|
||||
|
||||
self.cdm = ctx.obj.cdm
|
||||
self.drm_system = "playready" if is_playready_cdm(self.cdm) else "widevine"
|
||||
|
||||
self.user_profile = ctx.parent.params.get("profile")
|
||||
if not self.user_profile:
|
||||
self.user_profile = "default"
|
||||
|
||||
def get_session(self) -> session:
|
||||
return session()
|
||||
|
||||
def authenticate(self, cookies: CookieJar | None = None, credential: Credential | None = None) -> None:
|
||||
if not credential:
|
||||
raise EnvironmentError("Service requires Credentials for Authentication.")
|
||||
|
||||
self.session.headers.update(self.config["headers"])
|
||||
|
||||
cache = self.cache.get(f"tokens_{self.user_profile}_{credential.sha1}")
|
||||
|
||||
if cache and not cache.expired:
|
||||
# cached
|
||||
self.log.info(" + Using cached tokens...")
|
||||
profile = cache.data
|
||||
elif cache and cache.expired:
|
||||
# expired
|
||||
self.log.info(" + Authenticating...")
|
||||
profile = cache.data
|
||||
r = self.session.post(
|
||||
self.config["endpoints"]["login"],
|
||||
headers={"authorization": f"Basic {self.config['auth']['android']}"},
|
||||
data={
|
||||
"grant_type": "password",
|
||||
"username": credential.username,
|
||||
"password": credential.password,
|
||||
},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to acquire tokens: {r.text}")
|
||||
|
||||
tokens = r.json()
|
||||
profile["access_token"] = tokens["access_token"]
|
||||
profile["refresh_token"] = tokens["refresh_token"]
|
||||
cache.set(profile, expiration=tokens["expires_in"] - 30)
|
||||
|
||||
else:
|
||||
# new
|
||||
self.log.info(" + Authenticating...")
|
||||
r = self.session.post(
|
||||
self.config["endpoints"]["login"],
|
||||
headers={"authorization": f"Basic {self.config['auth']['android']}"},
|
||||
data={
|
||||
"grant_type": "password",
|
||||
"username": credential.username,
|
||||
"password": credential.password,
|
||||
},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to acquire tokens: {r.text}")
|
||||
|
||||
tokens = r.json()
|
||||
|
||||
account = self._get_account(tokens["access_token"])
|
||||
if not account.get("status", "").lower() == "active":
|
||||
raise ValueError("Account is not active. Please register first.")
|
||||
|
||||
account_id = account.get("id")
|
||||
profile = self._get_profile(account_id, tokens["access_token"])
|
||||
|
||||
profile_id = next((p.get("id") for p in profile if p.get("master") or p.get("maturity") == "ADULT"), None)
|
||||
|
||||
profile = {
|
||||
"access_token": tokens["access_token"],
|
||||
"refresh_token": tokens["refresh_token"],
|
||||
"account_id": account_id,
|
||||
"profile_id": profile_id,
|
||||
"profile_pin": "",
|
||||
}
|
||||
|
||||
cache.set(profile, expiration=tokens["expires_in"] - 30)
|
||||
|
||||
# Always refresh profile tokens
|
||||
self.log.info(" + Refreshing profile tokens...")
|
||||
r = self.session.post(
|
||||
self.config["endpoints"]["login"],
|
||||
headers={"authorization": f"Basic {self.config['auth']['android']}"},
|
||||
data={
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": profile["refresh_token"],
|
||||
"profile_id": profile["profile_id"],
|
||||
"profile_pin": "",
|
||||
},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to refresh tokens: {r.text}")
|
||||
|
||||
tokens = r.json()
|
||||
|
||||
data = {
|
||||
"platform": "platform_androidtv",
|
||||
"accessToken": f"{tokens['access_token']}"
|
||||
}
|
||||
|
||||
self.graphql_token = base64.b64encode(json.dumps(data).encode()).decode()
|
||||
self.jwt = tokens["access_token"]
|
||||
self.session.headers.update({"authorization": f"Bearer {self.jwt}"})
|
||||
|
||||
def get_titles(self):
|
||||
if not (match := re.match(self.TITLE_RE, self.title)):
|
||||
raise ValueError(f"Unable to parse title ID {self.title!r}")
|
||||
|
||||
entity_type, content_id = (match.group(i) for i in ("type", "id"))
|
||||
|
||||
if entity_type == "series":
|
||||
episodes = self._get_series(content_id)
|
||||
return Series(episodes)
|
||||
|
||||
elif entity_type == "movie":
|
||||
movie = self._get_movie(content_id)
|
||||
return Movies(movie)
|
||||
|
||||
elif entity_type == "play":
|
||||
episode = self._get_episode(content_id)
|
||||
return Series(episode)
|
||||
|
||||
def get_tracks(self, title: Movie | Episode) -> Tracks:
|
||||
r = self.session.get(
|
||||
url=self.config["endpoints"]["contents"].format(title.id),
|
||||
headers={
|
||||
"X-Client-Platform": "platform_jasper_androidtv",
|
||||
"X-Playback-Language": title.language.language,
|
||||
},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get content packages: {r.text}")
|
||||
data = r.json()
|
||||
|
||||
title.data["chapters"] = data.get("contentPackage", {}).get("breaks")
|
||||
|
||||
if not (package_id := data.get("contentPackage", {}).get("id")):
|
||||
raise ValueError(f"Failed to get package ID: {data}")
|
||||
if not (destination_id := data.get("contentPackage", {}).get("destinationId")):
|
||||
raise ValueError(f"Failed to get destination ID: {data}")
|
||||
|
||||
title.data["package_id"] = package_id
|
||||
title.data["destination_id"] = destination_id
|
||||
|
||||
params = {
|
||||
"format": "mpd",
|
||||
"filter": "ff",
|
||||
"uhd": "true",
|
||||
"hd": "true",
|
||||
"mcv": "true",
|
||||
"mca": "true",
|
||||
"mta": "true",
|
||||
"stt": "true",
|
||||
}
|
||||
|
||||
r = self.session.get(
|
||||
url=self.config["endpoints"]["destination"].format(title.id, package_id, destination_id),
|
||||
params=params,
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get content packages: {r.text}")
|
||||
data = r.json()
|
||||
|
||||
manifest, subtitles = self._get_assets(data)
|
||||
|
||||
tracks = DASH.from_url(manifest, self.session).to_tracks(title.language)
|
||||
|
||||
if subtitles is not None:
|
||||
tracks.add(
|
||||
Subtitle(
|
||||
id_=hashlib.md5(subtitles.encode()).hexdigest()[0:6],
|
||||
url=subtitles,
|
||||
codec=Subtitle.Codec.WebVTT,
|
||||
language=title.language,
|
||||
forced=False,
|
||||
sdh=True,
|
||||
)
|
||||
)
|
||||
|
||||
for track in tracks:
|
||||
if isinstance(track, Audio):
|
||||
role = track.data["dash"]["adaptation_set"].find("Role")
|
||||
if role is not None and role.get("value").lower() in ["description", "descriptive", "alternate"]:
|
||||
track.descriptive = True
|
||||
|
||||
track.is_original_lang = track.language == title.language
|
||||
track.name = "Original" if track.is_original_lang else None
|
||||
|
||||
elif isinstance(track, Subtitle):
|
||||
track.is_original_lang = track.language == title.language
|
||||
track.name = "Original" if track.is_original_lang else None
|
||||
if not track.forced:
|
||||
track.sdh = True
|
||||
|
||||
return tracks
|
||||
|
||||
def get_chapters(self, title: Episode | Movie) -> Chapters:
|
||||
if not (cues := title.data.get("chapters")):
|
||||
return Chapters()
|
||||
|
||||
chapters = []
|
||||
for chapter in cues:
|
||||
chapters.append(Chapter(name=chapter.get("breakType"), timestamp=int(chapter.get("startTime")) * 1000))
|
||||
if chapter.get("endTime"):
|
||||
chapters.append(Chapter(timestamp=int(chapter.get("endTime")) * 1000))
|
||||
|
||||
if not any(c.timestamp == "00:00:00.000" for c in chapters):
|
||||
chapters.append(Chapter(timestamp=0))
|
||||
|
||||
return sorted(chapters, key=lambda x: x.timestamp)
|
||||
|
||||
def get_widevine_service_certificate(self, challenge: bytes, **_: Any) -> bytes | None:
|
||||
if self.drm_system == "widevine":
|
||||
return self.session.post(url=self.license_url, data=challenge).content
|
||||
|
||||
return None
|
||||
|
||||
def get_widevine_license(self, *, challenge: bytes, title: Episode | Movie, track: Any, **kwargs) -> bytes | str | None:
|
||||
headers = {
|
||||
"Accept": "*/*",
|
||||
"Referer": "https://www.crave.ca/",
|
||||
"Origin": "https://www.crave.ca",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
data = {
|
||||
"payload": base64.b64encode(challenge).decode("utf-8"),
|
||||
"playbackContext": {
|
||||
"contentId": int(title.id),
|
||||
"contentpackageId": title.data["package_id"],
|
||||
"platformId": 1,
|
||||
"destinationId": title.data["destination_id"],
|
||||
"gl": "0",
|
||||
"jwt": self.jwt
|
||||
}
|
||||
}
|
||||
r = self.session.post(
|
||||
url=self.config["endpoints"]["widevine"],
|
||||
headers=headers,
|
||||
data=json.dumps(data),
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get license: {r.text}")
|
||||
|
||||
return r.content
|
||||
|
||||
def get_playready_license(self, *, challenge: str, title: Episode | Movie, track: Any, **kwargs) -> bytes | str | None:
|
||||
headers = {
|
||||
"Accept": "*/*",
|
||||
"Referer": "https://www.crave.ca/",
|
||||
"Origin": "https://www.crave.ca",
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
}
|
||||
data = {
|
||||
"payload": base64.b64encode(challenge.encode("utf-8")).decode("utf-8"),
|
||||
"playbackContext": {
|
||||
"contentId": int(title.id),
|
||||
"contentpackageId": title.data["package_id"],
|
||||
"platformId": 1,
|
||||
"destinationId": title.data["destination_id"],
|
||||
"gl": "0",
|
||||
"jwt": self.jwt
|
||||
}
|
||||
}
|
||||
r = self.session.post(
|
||||
url=self.config["endpoints"]["playready"],
|
||||
headers=headers,
|
||||
data=json.dumps(data),
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get license: {r.text}")
|
||||
|
||||
return r.content
|
||||
|
||||
# service-specific methods
|
||||
|
||||
def _get_assets(self, data: dict) -> tuple:
|
||||
manifest, trickplay = data.get("playback"), data.get("trickplay", "")
|
||||
if not manifest:
|
||||
raise ValueError(f"Failed to get manifest URL: {data}")
|
||||
|
||||
text_url = trickplay.replace("jpeg", "text")
|
||||
subtitles = text_url if self.session.head(text_url).ok else None
|
||||
|
||||
if not subtitles:
|
||||
self.log.warning("No external subtitle URL available")
|
||||
|
||||
# Shouldn't be needed with android auth, but just in case
|
||||
if "zultimate" not in manifest:
|
||||
manifest = re.sub(r"zbest-\w+", "zultimate-11110101", manifest)
|
||||
|
||||
if self.drm_system == "playready":
|
||||
manifest = manifest.replace("widevine", "playready")
|
||||
|
||||
return manifest, subtitles
|
||||
|
||||
def _get_account(self, access_token: str) -> dict:
|
||||
r = self.session.get(
|
||||
url=self.config["endpoints"]["account"],
|
||||
headers={"authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get account: {r.text}")
|
||||
return r.json()
|
||||
|
||||
def _get_profile(self, account_id: str, access_token: str) -> dict:
|
||||
r = self.session.get(
|
||||
url=self.config["endpoints"]["profile"].format(account_id),
|
||||
headers={"authorization": f"Bearer {access_token}"},
|
||||
)
|
||||
if not r.ok:
|
||||
raise ValueError(f"Failed to get profile: {r.text}")
|
||||
return r.json()
|
||||
|
||||
def _get_series(self, content_id: str) -> list[Episode]:
|
||||
payload = {
|
||||
"operationName": "GetShowpage",
|
||||
"variables": {
|
||||
"ids": [f"{content_id}"],
|
||||
"sessionContext": {"userMaturity": "ADULT", "userLanguage": "EN"}
|
||||
},
|
||||
"query": """
|
||||
query GetShowpage($sessionContext: SessionContext!, $ids: [String!]!) {
|
||||
medias(sessionContext: $sessionContext, ids: $ids) {
|
||||
id
|
||||
title
|
||||
mediaType
|
||||
productionYear
|
||||
originalLanguage {
|
||||
id
|
||||
displayName
|
||||
}
|
||||
seasons {
|
||||
id
|
||||
title
|
||||
seasonNumber
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
}
|
||||
res = self.session.post(
|
||||
url=self.config["endpoints"]["graphql"],
|
||||
headers={"authorization": f"Bearer {self.graphql_token}"},
|
||||
json=payload
|
||||
)
|
||||
if not res.ok:
|
||||
raise ConnectionError(f"Failed to request the title for {content_id}: {res.text}")
|
||||
data = res.json()
|
||||
|
||||
if not data["data"]["medias"]:
|
||||
raise ValueError(f"Failed to find the title for {content_id}")
|
||||
|
||||
original_language = data["data"]["medias"][0].get("originalLanguage", {}).get("id", "en")
|
||||
|
||||
seasons = data["data"]["medias"][0]["seasons"]
|
||||
|
||||
episodes = self._fetch_episodes(seasons)
|
||||
|
||||
return [
|
||||
Episode(
|
||||
id_=episode.get("id"),
|
||||
service=self.__class__,
|
||||
title=episode.get("media", {}).get("title"),
|
||||
year=episode.get("media", {}).get("productionYear"),
|
||||
season=int(episode.get("seasonNumber", 0)),
|
||||
number=int(episode.get("episodeNumber", 0)),
|
||||
name=episode.get("title"),
|
||||
language=original_language,
|
||||
data=episode,
|
||||
)
|
||||
for episode in episodes
|
||||
]
|
||||
|
||||
def _get_movie(self, content_id: str) -> list[Movie]:
|
||||
payload = {
|
||||
"operationName": "GetShowpage",
|
||||
"variables": {
|
||||
"ids": [f"{content_id}"],
|
||||
"sessionContext": {"userMaturity": "ADULT", "userLanguage": "EN"}
|
||||
},
|
||||
"query": """
|
||||
query GetShowpage($sessionContext: SessionContext!, $ids: [String!]!) {
|
||||
medias(sessionContext: $sessionContext, ids: $ids) {
|
||||
id
|
||||
title
|
||||
mediaType
|
||||
productionYear
|
||||
originalLanguage {
|
||||
id
|
||||
displayName
|
||||
}
|
||||
firstContent {
|
||||
id
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
}
|
||||
res = self.session.post(
|
||||
url=self.config["endpoints"]["graphql"],
|
||||
headers={"authorization": f"Bearer {self.graphql_token}"},
|
||||
json=payload
|
||||
)
|
||||
if not res.ok:
|
||||
raise ConnectionError(f"Failed to request the title for {content_id}: {res.text}")
|
||||
data = res.json()
|
||||
|
||||
if not data["data"]["medias"]:
|
||||
raise ValueError(f"Failed to find the title for {content_id}")
|
||||
|
||||
original_language = data["data"]["medias"][0].get("originalLanguage", {}).get("id", "en")
|
||||
title = data["data"]["medias"][0].get("title")
|
||||
|
||||
return [
|
||||
Movie(
|
||||
id_=data["data"]["medias"][0].get("firstContent", {}).get("id"),
|
||||
service=self.__class__,
|
||||
name=re.sub(r"\s\(\d{4}\)", "", title),
|
||||
year=data["data"]["medias"][0].get("productionYear"),
|
||||
language=original_language,
|
||||
data=data["data"]["medias"][0],
|
||||
)
|
||||
]
|
||||
|
||||
def _get_episode(self, content_id: str) -> json:
|
||||
payload = {
|
||||
"operationName": "GetPlayerPage",
|
||||
"variables": {"ids": [f"{content_id}"], "sessionContext": {"userMaturity": "ADULT", "userLanguage": "EN"}},
|
||||
"query": """
|
||||
query GetPlayerPage($sessionContext: SessionContext!, $ids: [String!]!) {
|
||||
contents(sessionContext: $sessionContext, ids: $ids) {
|
||||
id
|
||||
path
|
||||
title
|
||||
shortDescription
|
||||
seasonNumber
|
||||
episodeNumber
|
||||
contentType
|
||||
broadcastDate
|
||||
locked
|
||||
media {
|
||||
id
|
||||
title
|
||||
path
|
||||
mediaType
|
||||
productionYear
|
||||
originalLanguage {
|
||||
id
|
||||
displayName
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
""",
|
||||
}
|
||||
res = self.session.post(
|
||||
url=self.config["endpoints"]["graphql"],
|
||||
headers={"authorization": f"Bearer {self.graphql_token}"},
|
||||
json=payload
|
||||
)
|
||||
if not res.ok:
|
||||
raise ConnectionError(f"Failed to request the title for {content_id}: {res.text}")
|
||||
data = res.json()
|
||||
|
||||
if not (contents := data["data"]["contents"]):
|
||||
raise ValueError(f"Failed to find content for {content_id}")
|
||||
|
||||
original_language = contents[0].get("media", {}).get("originalLanguage", {}).get("id", "en")
|
||||
|
||||
return [
|
||||
Episode(
|
||||
id_=content.get("id"),
|
||||
service=self.__class__,
|
||||
title=content.get("media", {}).get("title"),
|
||||
year=content.get("media", {}).get("productionYear"),
|
||||
season=int(content.get("seasonNumber", 0)),
|
||||
number=int(content.get("episodeNumber", 0)),
|
||||
name=content.get("title"),
|
||||
language=original_language,
|
||||
data=content,
|
||||
)
|
||||
for content in contents
|
||||
]
|
||||
|
||||
def _fetch_episode(self, season_id: str) -> json:
|
||||
payload = {
|
||||
"operationName": "GetContentBySeasonId",
|
||||
"variables": {
|
||||
"id": season_id,
|
||||
"sessionContext": {"userMaturity": "ADULT", "userLanguage": "EN"},
|
||||
"contentFormat": {"format": "LONGFORM"},
|
||||
},
|
||||
"query": """
|
||||
query GetContentBySeasonId($sessionContext: SessionContext!, $id: String!, $contentFormat: ContentFormatRequest) {
|
||||
contentsBySeasonId(
|
||||
sessionContext: $sessionContext
|
||||
id: $id
|
||||
contentFormat: $contentFormat
|
||||
) {
|
||||
id
|
||||
locked
|
||||
title
|
||||
episodeNumber
|
||||
seasonNumber
|
||||
broadcastDate
|
||||
path
|
||||
shortDescription
|
||||
languageIndicators {
|
||||
indicator
|
||||
languages {
|
||||
langCode
|
||||
displayTitle
|
||||
locked
|
||||
}
|
||||
}
|
||||
media {
|
||||
id
|
||||
title
|
||||
path
|
||||
productionYear
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
}
|
||||
response = self.session.post(
|
||||
url=self.config["endpoints"]["graphql"],
|
||||
headers={"authorization": f"Bearer {self.graphql_token}"},
|
||||
json=payload
|
||||
)
|
||||
if not response.ok:
|
||||
raise ConnectionError(f"Failed to request the title for {season_id}: {response.text}")
|
||||
data = response.json()
|
||||
return data["data"]["contentsBySeasonId"]
|
||||
|
||||
def _fetch_episodes(self, data: dict) -> list:
|
||||
with ThreadPoolExecutor(max_workers=10) as executor:
|
||||
tasks = [executor.submit(self._fetch_episode, x["id"]) for x in data]
|
||||
titles = [future.result() for future in as_completed(tasks)]
|
||||
return [episode for episodes in titles for episode in episodes]
|
||||
@@ -0,0 +1,16 @@
|
||||
headers:
|
||||
user-agent: "Dalvik/2.1.0 (Linux; U; Android 11; SHIELD Android TV Build/RQ1A.210105.003)"
|
||||
|
||||
endpoints:
|
||||
graphql: "https://rte-api.bellmedia.ca/graphql"
|
||||
login: "https://account.bellmedia.ca/api/login/v2.2"
|
||||
account: "https://account.bellmedia.ca/api/account/v1.1"
|
||||
profile: "https://account.bellmedia.ca/api/profile/v2/account/{}"
|
||||
contents: "https://playback.rte-api.bellmedia.ca/contents/{}"
|
||||
destination: "https://stream.video.9c9media.com/meta/content/{}/contentpackage/{}/destination/{}/platform/1"
|
||||
widevine: "https://license.9c9media.com/widevine"
|
||||
playready: "https://license.9c9media.com/playready"
|
||||
|
||||
auth:
|
||||
android: "Y3JhdmUtYW5kcm9pZDpkZWZhdWx0"
|
||||
web: "Y3JhdmUtd2ViOmRlZmF1bHQ="
|
||||
@@ -1,407 +1,396 @@
|
||||
import base64
|
||||
import json
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from http.cookiejar import CookieJar
|
||||
from typing import List, Optional
|
||||
from collections.abc import Generator
|
||||
from http.cookiejar import MozillaCookieJar
|
||||
from typing import Any, Optional
|
||||
|
||||
import click
|
||||
import jwt
|
||||
from click import Context
|
||||
from langcodes import Language
|
||||
|
||||
from envied.core.constants import AnyTrack
|
||||
from envied.core.credential import Credential
|
||||
from envied.core.manifests import DASH
|
||||
from envied.core.manifests import DASH, HLS
|
||||
from envied.core.search_result import SearchResult
|
||||
from envied.core.service import Service
|
||||
from envied.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
|
||||
from envied.core.tracks import Subtitle, Tracks
|
||||
from envied.core.titles import Episode, Movie, Movies, Series
|
||||
from envied.core.tracks import Chapters, Subtitle, Tracks
|
||||
|
||||
|
||||
class KNPY(Service):
|
||||
"""
|
||||
Service code for Kanopy (kanopy.com).
|
||||
Version: 1.0.0
|
||||
\b
|
||||
Service code for Kanopy streaming service (https://www.kanopy.com/).
|
||||
|
||||
Auth: Credential (username + password)
|
||||
Security: FHD@L3
|
||||
\b
|
||||
Version: 1.0.1
|
||||
Author: stabbedbybrick
|
||||
Authorization: Credentials
|
||||
Geofence: None
|
||||
Robustness:
|
||||
widevine:
|
||||
L3: 1080p
|
||||
playready:
|
||||
SL2000: 1080p
|
||||
|
||||
\b
|
||||
Tips:
|
||||
- Input can be complete title URL or video path:
|
||||
MOVIE: https://www.kanopy.com/en/lapl/video/16239510 OR /video/16239510 OR /product/16239510
|
||||
SERIES: https://www.kanopy.com/en/lapl/video/14949910 OR /video/14949910 OR /product/14949910
|
||||
EPISODE: https://www.kanopy.com/en/lapl/video/14949910/14949912 OR /video/14949910/14949912
|
||||
|
||||
\b
|
||||
Notes:
|
||||
- Available tickets are checked on each run, but they don't appear to be used
|
||||
when titles are accessed through the API.
|
||||
|
||||
Handles both Movies and Series (Playlists).
|
||||
Detects and stops for movies that require tickets.
|
||||
Caching included
|
||||
"""
|
||||
|
||||
# Updated regex to match the new URL structure with library subdomain and path
|
||||
TITLE_RE = r"^https?://(?:www\.)?kanopy\.com/.+/(?P<id>\d+)$"
|
||||
GEOFENCE = ()
|
||||
NO_SUBTITLES = False
|
||||
# GEOFENCE = ()
|
||||
ALIASES = ("kanopy",)
|
||||
TITLE_RE = r"^(?:.*?/)?(?P<type>watch/video|video|watch|product)/(?P<id1>\d+)(?:/(?P<id2>\d+))?/?$"
|
||||
|
||||
@staticmethod
|
||||
@click.command(name="KNPY", short_help="https://kanopy.com")
|
||||
@click.command(name="KNPY", short_help="https://www.kanopy.com/", help=__doc__)
|
||||
@click.argument("title", type=str)
|
||||
@click.pass_context
|
||||
def cli(ctx, **kwargs):
|
||||
def cli(ctx: Context, **kwargs: Any) -> KNPY:
|
||||
return KNPY(ctx, **kwargs)
|
||||
|
||||
def __init__(self, ctx, title: str):
|
||||
def __init__(self, ctx: Context, title: str):
|
||||
self.title = title
|
||||
super().__init__(ctx)
|
||||
if not self.config:
|
||||
raise ValueError("KNPY configuration not found. Ensure config.yaml exists.")
|
||||
|
||||
self.cdm = ctx.obj.cdm
|
||||
self.session.headers.update(self.config["headers"])
|
||||
|
||||
match = re.match(self.TITLE_RE, title)
|
||||
if match:
|
||||
self.content_id = match.group("id")
|
||||
else:
|
||||
self.content_id = None
|
||||
self.search_query = title
|
||||
def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
|
||||
super().authenticate(cookies, credential)
|
||||
if not credential:
|
||||
raise EnvironmentError("Service requires Credentials for Authentication.")
|
||||
|
||||
self.API_VERSION = self.config["client"]["api_version"]
|
||||
self.USER_AGENT = self.config["client"]["user_agent"]
|
||||
self.WIDEVINE_UA = self.config["client"]["widevine_ua"]
|
||||
|
||||
self.session.headers.update({
|
||||
"x-version": self.API_VERSION,
|
||||
"user-agent": self.USER_AGENT
|
||||
})
|
||||
|
||||
self._jwt = None
|
||||
self._visitor_id = None
|
||||
self._user_id = None
|
||||
self._domain_id = None
|
||||
self.widevine_license_url = None
|
||||
|
||||
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
|
||||
if not credential or not credential.username or not credential.password:
|
||||
raise ValueError("Kanopy requires email and password for authentication.")
|
||||
|
||||
cache = self.cache.get("auth_token")
|
||||
cache = self.cache.get(f"tokens_{credential.sha1}")
|
||||
|
||||
if cache and not cache.expired:
|
||||
cached_data = cache.data
|
||||
valid_token = None
|
||||
|
||||
if isinstance(cached_data, dict) and "token" in cached_data:
|
||||
if cached_data.get("username") == credential.username:
|
||||
valid_token = cached_data["token"]
|
||||
self.log.info("Using cached authentication token")
|
||||
else:
|
||||
self.log.info(f"Cached token belongs to '{cached_data.get('username')}', but logging in as '{credential.username}'. Re-authenticating.")
|
||||
|
||||
elif isinstance(cached_data, str):
|
||||
self.log.info("Found legacy cached token format. Re-authenticating to ensure correct user.")
|
||||
|
||||
if valid_token:
|
||||
self._jwt = valid_token
|
||||
self.session.headers.update({"authorization": f"Bearer {self._jwt}"})
|
||||
|
||||
if not self._user_id or not self._domain_id or not self._visitor_id:
|
||||
try:
|
||||
decoded_jwt = jwt.decode(self._jwt, options={"verify_signature": False})
|
||||
self._user_id = decoded_jwt["data"]["uid"]
|
||||
self._visitor_id = decoded_jwt["data"]["visitor_id"]
|
||||
self.log.info(f"Extracted user_id and visitor_id from cached token.")
|
||||
self._fetch_user_details()
|
||||
return
|
||||
except (KeyError, jwt.DecodeError) as e:
|
||||
self.log.error(f"Could not decode cached token: {e}. Re-authenticating.")
|
||||
|
||||
self.log.info("Performing handshake to get visitor token...")
|
||||
r = self.session.get(self.config["endpoints"]["handshake"])
|
||||
r.raise_for_status()
|
||||
handshake_data = r.json()
|
||||
self._visitor_id = handshake_data["visitorId"]
|
||||
initial_jwt = handshake_data["jwt"]
|
||||
|
||||
self.log.info(f"Logging in as {credential.username}...")
|
||||
login_payload = {
|
||||
"credentialType": "email",
|
||||
"emailUser": {
|
||||
"email": credential.username,
|
||||
"password": credential.password
|
||||
}
|
||||
}
|
||||
r = self.session.post(
|
||||
self.config["endpoints"]["login"],
|
||||
json=login_payload,
|
||||
headers={"authorization": f"Bearer {initial_jwt}"}
|
||||
)
|
||||
r.raise_for_status()
|
||||
login_data = r.json()
|
||||
self._jwt = login_data["jwt"]
|
||||
self._user_id = login_data["userId"]
|
||||
|
||||
self.session.headers.update({"authorization": f"Bearer {self._jwt}"})
|
||||
self.log.info(f"Successfully authenticated as {credential.username}")
|
||||
|
||||
self._fetch_user_details()
|
||||
|
||||
try:
|
||||
decoded_jwt = jwt.decode(self._jwt, options={"verify_signature": False})
|
||||
exp_timestamp = decoded_jwt.get("exp")
|
||||
|
||||
cache_payload = {
|
||||
"token": self._jwt,
|
||||
"username": credential.username
|
||||
}
|
||||
|
||||
if exp_timestamp:
|
||||
expiration_in_seconds = int(exp_timestamp - datetime.now(timezone.utc).timestamp())
|
||||
self.log.info(f"Caching token for {expiration_in_seconds / 60:.2f} minutes.")
|
||||
cache.set(data=cache_payload, expiration=expiration_in_seconds)
|
||||
else:
|
||||
self.log.warning("JWT has no 'exp' claim, caching for 1 hour as a fallback.")
|
||||
cache.set(data=cache_payload, expiration=3600)
|
||||
except Exception as e:
|
||||
self.log.error(f"Failed to decode JWT for caching: {e}. Caching for 1 hour as a fallback.")
|
||||
cache.set(
|
||||
data={"token": self._jwt, "username": credential.username},
|
||||
expiration=3600
|
||||
)
|
||||
|
||||
def _fetch_user_details(self):
|
||||
self.log.info("Fetching user library memberships...")
|
||||
r = self.session.get(self.config["endpoints"]["memberships"].format(user_id=self._user_id))
|
||||
r.raise_for_status()
|
||||
memberships = r.json()
|
||||
|
||||
for membership in memberships.get("list", []):
|
||||
if membership.get("status") == "active" and membership.get("isDefault", False):
|
||||
self._domain_id = str(membership["domainId"])
|
||||
self.log.info(f"Using default library domain: {membership.get('sitename', 'Unknown')} (ID: {self._domain_id})")
|
||||
return
|
||||
|
||||
if memberships.get("list"):
|
||||
self._domain_id = str(memberships["list"][0]["domainId"])
|
||||
self.log.warning(f"No default library found. Using first active domain: {self._domain_id}")
|
||||
self.log.info(" + Using cached tokens")
|
||||
tokens = cache.data
|
||||
else:
|
||||
raise ValueError("No active library memberships found for this user.")
|
||||
self.log.info(" + Logging in...")
|
||||
payload = {
|
||||
"credentialType": "email",
|
||||
"emailUser": {
|
||||
"email": credential.username,
|
||||
"password": credential.password,
|
||||
},
|
||||
}
|
||||
|
||||
def get_titles(self) -> Titles_T:
|
||||
if not self.content_id:
|
||||
raise ValueError("A content ID is required to get titles. Use a URL or run a search first.")
|
||||
if not self._domain_id:
|
||||
raise ValueError("Domain ID not set. Authentication may have failed.")
|
||||
|
||||
r = self.session.get(self.config["endpoints"]["video_info"].format(video_id=self.content_id, domain_id=self._domain_id))
|
||||
r.raise_for_status()
|
||||
content_data = r.json()
|
||||
|
||||
content_type = content_data.get("type")
|
||||
|
||||
def parse_lang(data):
|
||||
try:
|
||||
langs = data.get("languages", [])
|
||||
if langs and isinstance(langs, list) and len(langs) > 0:
|
||||
return Language.find(langs[0])
|
||||
except:
|
||||
pass
|
||||
return Language.get("en")
|
||||
|
||||
if content_type == "video":
|
||||
video_data = content_data["video"]
|
||||
movie = Movie(
|
||||
id_=str(video_data["videoId"]),
|
||||
service=self.__class__,
|
||||
name=video_data["title"],
|
||||
year=video_data.get("productionYear"),
|
||||
description=video_data.get("descriptionHtml", ""),
|
||||
language=parse_lang(video_data),
|
||||
data=video_data,
|
||||
response = self.session.post(
|
||||
self.config["endpoints"]["login"], json=payload
|
||||
)
|
||||
return Movies([movie])
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
elif content_type == "playlist":
|
||||
playlist_data = content_data["playlist"]
|
||||
series_title = playlist_data["title"]
|
||||
series_year = playlist_data.get("productionYear")
|
||||
if not (access_token := data.get("jwt")):
|
||||
raise ConnectionError(f"Failed to login: {data}")
|
||||
|
||||
season_match = re.search(r'(?:Season|S)\s*(\d+)', series_title, re.IGNORECASE)
|
||||
season_num = int(season_match.group(1)) if season_match else 1
|
||||
user_id = data.get("userId")
|
||||
|
||||
r = self.session.get(self.config["endpoints"]["video_items"].format(video_id=self.content_id, domain_id=self._domain_id))
|
||||
r.raise_for_status()
|
||||
items_data = r.json()
|
||||
expiry = jwt.decode(access_token, options={"verify_signature": False}).get("exp")
|
||||
if not expiry:
|
||||
expiry = 86400 # 24 hour fallback
|
||||
|
||||
episodes = []
|
||||
for i, item in enumerate(items_data.get("list", [])):
|
||||
if item.get("type") != "video":
|
||||
tokens = {
|
||||
"accessToken": access_token,
|
||||
"userId": user_id,
|
||||
}
|
||||
|
||||
cache.set(tokens, expiration=expiry - 3600)
|
||||
|
||||
user = self._get_membership(tokens.get("accessToken"), tokens.get("userId"))
|
||||
|
||||
self.subdomain = user.get("subdomain", "lapl")
|
||||
self.site = user.get("sitename")
|
||||
self.domain_id = user.get("domainId")
|
||||
self.available_tickets = user.get("ticketsAvailable", 0)
|
||||
|
||||
self.log.info(f"Site: {self.site} | Available tickets: {self.available_tickets}")
|
||||
|
||||
self.user_id = tokens.get("userId")
|
||||
self.access_token = tokens.get("accessToken")
|
||||
self.session.headers.update({"authorization": "Bearer {}".format(self.access_token)})
|
||||
|
||||
def search(self) -> Generator[SearchResult, None, None]:
|
||||
params = {
|
||||
"query": self.title,
|
||||
"sort": "relevance",
|
||||
"rfp": "exclude",
|
||||
"domainId": self.domain_id,
|
||||
"isKids": "false",
|
||||
"page": "0",
|
||||
"perPage": "40",
|
||||
}
|
||||
|
||||
response = self.session.get(self.config["endpoints"]["search"], params=params)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
for result in data.get("list", []):
|
||||
yield SearchResult(
|
||||
id_=f"/video/{result.get("videoId")}",
|
||||
title=result.get("title"),
|
||||
description=result.get("tagline"),
|
||||
label="Video",
|
||||
url=f"/video/{result.get("videoId")}",
|
||||
)
|
||||
|
||||
def get_titles(self) -> Movies | Series:
|
||||
match = re.match(self.TITLE_RE, self.title)
|
||||
if not match:
|
||||
raise ValueError(f"Invalid title: {self.title}")
|
||||
|
||||
_, content_id, video_id = (match.group(i) for i in ("type", "id1", "id2"))
|
||||
|
||||
response = self.session.get(
|
||||
self.config["endpoints"]["videos"].format(video_id=content_id),
|
||||
params={"domainId": self.domain_id, "ageRatingDomainId": self.domain_id},
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
entity_type = data.get("type")
|
||||
|
||||
if entity_type in ("video"):
|
||||
movie = self._get_movie(data)
|
||||
return Movies(movie)
|
||||
|
||||
# listed as a single season
|
||||
elif entity_type in ("playlist"):
|
||||
episodes = self._get_playlist(data)
|
||||
if video_id is not None:
|
||||
episodes = [episode for episode in episodes if int(episode.id) == int(video_id)]
|
||||
return Series(episodes)
|
||||
|
||||
# listed as a collection of seasons
|
||||
elif entity_type in ("collection"):
|
||||
episodes = self._get_collection(data)
|
||||
if video_id is not None:
|
||||
episodes = [episode for episode in episodes if int(episode.id) == int(video_id)]
|
||||
return Series(episodes)
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unsupported content type: {entity_type}")
|
||||
|
||||
def get_tracks(self, title: Movie | Episode) -> Tracks:
|
||||
json_data = {
|
||||
"videoId": title.id,
|
||||
"userId": self.user_id,
|
||||
"domainId": self.domain_id,
|
||||
}
|
||||
|
||||
response = self.session.post(
|
||||
self.config["endpoints"]["plays"],
|
||||
headers={
|
||||
"authorization": "Bearer {}".format(self.access_token),
|
||||
**self.config["headers"],
|
||||
},
|
||||
json=json_data,
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
manifests = data.get("manifests", [])
|
||||
|
||||
stream_data = next((m for m in manifests if m.get("manifestType") == "dash"), None)
|
||||
stream_format = DASH
|
||||
|
||||
# Fallback to HLS
|
||||
if not stream_data:
|
||||
stream_data = next((m for m in manifests if m.get("manifestType") == "hls"), None)
|
||||
stream_format = HLS
|
||||
|
||||
if not stream_data:
|
||||
raise ValueError(f"Failed to find manifest for title: {title}")
|
||||
|
||||
manifest = stream_data.get("url")
|
||||
title.data["drm_id"] = stream_data.get("drmLicenseID")
|
||||
|
||||
self.session.headers.clear()
|
||||
tracks = stream_format.from_url(manifest, self.session).to_tracks(title.language)
|
||||
|
||||
for caption in data.get("captions", []):
|
||||
language = caption.get("language")
|
||||
for sub in caption.get("files", []):
|
||||
if sub.get("type") == "transcript":
|
||||
continue
|
||||
|
||||
video_data = item["video"]
|
||||
ep_num = i + 1
|
||||
|
||||
ep_title = video_data.get("title", "")
|
||||
ep_match = re.search(r'Ep(?:isode)?\.?\s*(\d+)', ep_title, re.IGNORECASE)
|
||||
if ep_match:
|
||||
ep_num = int(ep_match.group(1))
|
||||
|
||||
episodes.append(
|
||||
Episode(
|
||||
id_=str(video_data["videoId"]),
|
||||
service=self.__class__,
|
||||
title=series_title,
|
||||
season=season_num,
|
||||
number=ep_num,
|
||||
name=video_data["title"],
|
||||
description=video_data.get("descriptionHtml", ""),
|
||||
year=video_data.get("productionYear", series_year),
|
||||
language=parse_lang(video_data),
|
||||
data=video_data,
|
||||
tracks.add(
|
||||
Subtitle(
|
||||
id_=hashlib.md5(sub.get("url").encode()).hexdigest()[0:6],
|
||||
codec=Subtitle.Codec.from_codecs(sub.get("url").split(".")[-1]),
|
||||
language=language,
|
||||
url=sub.get("url"),
|
||||
)
|
||||
)
|
||||
|
||||
series = Series(episodes)
|
||||
series.name = series_title
|
||||
series.description = playlist_data.get("descriptionHtml", "")
|
||||
series.year = series_year
|
||||
return series
|
||||
|
||||
else:
|
||||
raise ValueError(f"Unsupported content type: {content_type}")
|
||||
|
||||
def get_tracks(self, title: Title_T) -> Tracks:
|
||||
play_payload = {
|
||||
"videoId": int(title.id),
|
||||
"domainId": int(self._domain_id),
|
||||
"userId": int(self._user_id),
|
||||
"visitorId": self._visitor_id
|
||||
}
|
||||
|
||||
self.session.headers.setdefault("authorization", f"Bearer {self._jwt}")
|
||||
self.session.headers.setdefault("x-version", self.API_VERSION)
|
||||
self.session.headers.setdefault("user-agent", self.USER_AGENT)
|
||||
|
||||
r = self.session.post(self.config["endpoints"]["plays"], json=play_payload)
|
||||
response_json = None
|
||||
try:
|
||||
response_json = r.json()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Handle known errors gracefully
|
||||
if r.status_code == 403:
|
||||
if response_json and response_json.get("errorSubcode") == "playRegionRestricted":
|
||||
self.log.error("Kanopy reports: This video is not available in your country.")
|
||||
raise PermissionError(
|
||||
"Playback blocked by region restriction. Try connecting through a supported country or verify your library’s access region."
|
||||
)
|
||||
else:
|
||||
self.log.error(f"Access forbidden (HTTP 403). Response: {response_json}")
|
||||
raise PermissionError("Kanopy denied access to this video. It may require a different library membership or authentication.")
|
||||
|
||||
# Raise for any other HTTP errors
|
||||
r.raise_for_status()
|
||||
play_data = response_json or r.json()
|
||||
|
||||
manifest_url = None
|
||||
for manifest in play_data.get("manifests", []):
|
||||
if manifest["manifestType"] == "dash":
|
||||
url = manifest["url"]
|
||||
manifest_url = f"https://kanopy.com{url}" if url.startswith("/") else url
|
||||
drm_type = manifest.get("drmType")
|
||||
if drm_type == "kanopyDrm":
|
||||
play_id = play_data.get("playId")
|
||||
self.widevine_license_url = self.config["endpoints"]["widevine_license"].format(license_id=f"{play_id}-0")
|
||||
elif drm_type == "studioDrm":
|
||||
license_id = manifest.get("drmLicenseID", f"{play_data.get('playId')}-1")
|
||||
self.widevine_license_url = self.config["endpoints"]["widevine_license"].format(license_id=license_id)
|
||||
else:
|
||||
self.log.warning(f"Unknown drmType: {drm_type}")
|
||||
self.widevine_license_url = None
|
||||
break
|
||||
|
||||
if not manifest_url:
|
||||
raise ValueError("Could not find a DASH manifest for this title.")
|
||||
if not self.widevine_license_url:
|
||||
raise ValueError("Could not construct Widevine license URL.")
|
||||
|
||||
self.log.info(f"Fetching DASH manifest from: {manifest_url}")
|
||||
r = self.session.get(manifest_url)
|
||||
r.raise_for_status()
|
||||
|
||||
# Refresh headers for manifest parsing
|
||||
self.session.headers.clear()
|
||||
self.session.headers.update({
|
||||
"User-Agent": self.WIDEVINE_UA,
|
||||
"Accept": "*/*",
|
||||
"Accept-Encoding": "gzip, deflate",
|
||||
"Connection": "keep-alive",
|
||||
})
|
||||
|
||||
tracks = DASH.from_text(r.text, url=manifest_url).to_tracks(language=title.language)
|
||||
for caption_data in play_data.get("captions", []):
|
||||
lang = caption_data.get("language", "en")
|
||||
for file_info in caption_data.get("files", []):
|
||||
if file_info.get("type") == "webvtt":
|
||||
tracks.add(Subtitle(
|
||||
id_=f"caption-{lang}",
|
||||
url=file_info["url"],
|
||||
codec=Subtitle.Codec.WebVTT,
|
||||
language=Language.get(lang)
|
||||
))
|
||||
break
|
||||
|
||||
return tracks
|
||||
|
||||
def get_chapters(self, title: Movie | Episode) -> Chapters:
|
||||
return Chapters()
|
||||
|
||||
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes:
|
||||
if not self.widevine_license_url:
|
||||
raise ValueError("Widevine license URL was not set. Call get_tracks first.")
|
||||
def get_widevine_service_certificate(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
|
||||
return None
|
||||
|
||||
license_headers = {
|
||||
"Content-Type": "application/octet-stream",
|
||||
"User-Agent": self.WIDEVINE_UA,
|
||||
"Authorization": f"Bearer {self._jwt}",
|
||||
"X-Version": self.API_VERSION
|
||||
def get_widevine_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
|
||||
if not (drm_id := title.data.get("drm_id")):
|
||||
return None
|
||||
|
||||
license_url = self.config["endpoints"]["widevine"].format(drm_id=drm_id)
|
||||
|
||||
headers = {
|
||||
"authorization": "Bearer {}".format(self.access_token),
|
||||
"content-type": "application/octet-stream",
|
||||
**self.config["headers"]
|
||||
}
|
||||
|
||||
r = self.session.post(
|
||||
self.widevine_license_url,
|
||||
data=challenge,
|
||||
headers=license_headers
|
||||
)
|
||||
r = self.session.post(url=license_url, headers=headers, data=challenge)
|
||||
r.raise_for_status()
|
||||
|
||||
return r.content
|
||||
|
||||
# def search(self) -> List[SearchResult]:
|
||||
# if not hasattr(self, 'search_query'):
|
||||
# self.log.error("Search query not set. Cannot search.")
|
||||
# return []
|
||||
def get_playready_license(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None:
|
||||
if not (drm_id := title.data.get("drm_id")):
|
||||
return None
|
||||
|
||||
# self.log.info(f"Searching for '{self.search_query}'...")
|
||||
# params = {
|
||||
# "query": self.search_query,
|
||||
# "sort": "relevance",
|
||||
# "domainId": self._domain_id,
|
||||
# "page": 0,
|
||||
# "perPage": 20
|
||||
# }
|
||||
# r = self.session.get(self.config["endpoints"]["search"], params=params)
|
||||
# r.raise_for_status()
|
||||
# search_data = r.json()
|
||||
license_url = self.config["endpoints"]["playready"].format(drm_id=drm_id)
|
||||
|
||||
# results = []
|
||||
# for item in search_data.get("list", []):
|
||||
# item_type = item.get("type")
|
||||
# if item_type not in ["playlist", "video"]:
|
||||
# continue
|
||||
headers = {
|
||||
"authorization": "Bearer {}".format(self.access_token),
|
||||
"content-type": "text/xml; charset=utf-8",
|
||||
"SOAPAction": "http://schemas.microsoft.com/DRM/2007/03/protocols/AcquireLicense",
|
||||
**self.config["headers"]
|
||||
}
|
||||
|
||||
# video_id = item.get("videoId")
|
||||
# title = item.get("title", "No Title")
|
||||
# label = "Series" if item_type == "playlist" else "Movie"
|
||||
r = self.session.post(url=license_url, headers=headers, data=challenge)
|
||||
r.raise_for_status()
|
||||
|
||||
return r.content
|
||||
|
||||
# Service-specific methods
|
||||
|
||||
def _get_playlist(self, data: dict) -> list[Episode]:
|
||||
if not (metadata := data.get("playlist")):
|
||||
self.log.error(" - Error: No metadata found for this title.")
|
||||
return []
|
||||
|
||||
series_title = metadata.get("title")
|
||||
season = re.search(r"\b(?:S|Season|Series)\s*(\d+)", series_title, re.IGNORECASE)
|
||||
season_number = int(season.group(1)) if season else 1
|
||||
|
||||
response = self.session.get(
|
||||
self.config["endpoints"]["items"].format(video_id=metadata.get("videoId")),
|
||||
params={"domainId": self.domain_id, "ageRatingDomainId": self.domain_id}
|
||||
)
|
||||
if response.status_code != 200:
|
||||
self.log.error(f" - Error: Failed to fetch playlist for ID: {metadata.get('videoId')} - {response.status_code}")
|
||||
return []
|
||||
|
||||
data = response.json()
|
||||
|
||||
videos = []
|
||||
for e, item in enumerate(data.get("list", [])):
|
||||
if item.get("type", "").lower() != "video":
|
||||
continue
|
||||
|
||||
episode = item.get("video", {})
|
||||
if episode:
|
||||
episode["number"] = e + 1
|
||||
videos.append(episode)
|
||||
|
||||
episodes = []
|
||||
for video in videos:
|
||||
lang = next((x.get("name") for x in video.get("taxonomies", {}).get("languages", [])), "English")
|
||||
episodes.append(
|
||||
Episode(
|
||||
id_=video.get("videoId"),
|
||||
service=self.__class__,
|
||||
title=series_title,
|
||||
season=season_number,
|
||||
number=video.get("number", 0),
|
||||
name=video.get("title"),
|
||||
year=video.get("productionYear"),
|
||||
language=Language.find(lang).to_alpha3(),
|
||||
data=video,
|
||||
)
|
||||
)
|
||||
|
||||
return episodes
|
||||
|
||||
def _get_collection(self, data: dict) -> list[Episode]:
|
||||
if not (metadata := data.get("collection")):
|
||||
self.log.error(" - Error: No metadata found for this title.")
|
||||
return []
|
||||
|
||||
series_title = metadata.get("title")
|
||||
|
||||
response = self.session.get(
|
||||
self.config["endpoints"]["items"].format(video_id=metadata.get("videoId")),
|
||||
params={"domainId": self.domain_id, "ageRatingDomainId": self.domain_id}
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
playlists = [i for i in data.get("list", []) if i.get("type", "").lower() == "playlist"]
|
||||
if not playlists:
|
||||
self.log.error(" - Error: No playlists found for this title.")
|
||||
return []
|
||||
|
||||
episodes = []
|
||||
for s, playlist in enumerate(playlists):
|
||||
playlist_episodes = self._get_playlist(playlist)
|
||||
|
||||
for video in playlist_episodes:
|
||||
video.title = series_title
|
||||
video.season = video.season or s + 1
|
||||
|
||||
episodes.extend(playlist_episodes)
|
||||
|
||||
return episodes
|
||||
|
||||
def _get_movie(self, data: dict) -> list[Movie]:
|
||||
if not (metadata := data.get("video")):
|
||||
self.log.error(" - Error: No metadata found for this title.")
|
||||
return []
|
||||
|
||||
lang = next((x.get("name") for x in metadata.get("languages", [])), "English")
|
||||
|
||||
return [
|
||||
Movie(
|
||||
id_=metadata.get("videoId"),
|
||||
service=self.__class__,
|
||||
name=metadata.get("title"),
|
||||
year=metadata.get("productionYear"),
|
||||
language=Language.find(lang).to_alpha3(),
|
||||
data=metadata,
|
||||
)
|
||||
]
|
||||
|
||||
def _get_membership(self, access_token: str, user_id: str) -> dict:
|
||||
response = self.session.get(
|
||||
self.config["endpoints"]["memberships"],
|
||||
headers={"authorization": "Bearer {}".format(access_token)},
|
||||
params={"userId": user_id},
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
|
||||
user_info = next((i for i in data.get("list", []) if i.get("userId") == user_id), None)
|
||||
if not user_info:
|
||||
raise ValueError(f"Failed to get membership info for user: {data}")
|
||||
|
||||
if user_info.get("status", "").lower() != "active":
|
||||
raise ValueError(f"User is not active: {user_info}")
|
||||
|
||||
return user_info
|
||||
|
||||
# results.append(
|
||||
# SearchResult(
|
||||
# id_=str(video_id),
|
||||
# title=title,
|
||||
# description="",
|
||||
# label=label,
|
||||
# url=f"https://www.kanopy.com/watch/{video_id}"
|
||||
# )
|
||||
# )
|
||||
# return results
|
||||
|
||||
def get_chapters(self, title: Title_T) -> list:
|
||||
return []
|
||||
|
||||
@@ -1,15 +1,15 @@
|
||||
client:
|
||||
api_version: "Android/com.kanopy/6.21.0/952 (SM-A525F; Android 15)"
|
||||
user_agent: "okhttp/5.2.1"
|
||||
widevine_ua: "KanopyApplication/6.21.0 (Linux;Android 15) AndroidXMedia3/1.8.0"
|
||||
headers:
|
||||
user-agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/147.0.0.0 Safari/537.36"
|
||||
x-version: "web/undefined/undefined/undefined"
|
||||
|
||||
endpoints:
|
||||
handshake: "https://kanopy.com/kapi/handshake"
|
||||
login: "https://kanopy.com/kapi/login"
|
||||
memberships: "https://kanopy.com/kapi/memberships?userId={user_id}"
|
||||
video_info: "https://kanopy.com/kapi/videos/{video_id}?domainId={domain_id}"
|
||||
video_items: "https://kanopy.com/kapi/videos/{video_id}/items?domainId={domain_id}"
|
||||
search: "https://kanopy.com/kapi/search/videos"
|
||||
plays: "https://kanopy.com/kapi/plays"
|
||||
access_expires_in: "https://kanopy.com/kapi/users/{user_id}/history/videos/{video_id}/access_expires_in?domainId={domain_id}"
|
||||
widevine_license: "https://kanopy.com/kapi/licenses/widevine/{license_id}"
|
||||
login: "https://www.kanopy.com/kapi/login"
|
||||
users: "https://www.kanopy.com/kapi/users/{user_id}"
|
||||
memberships: "https://www.kanopy.com/kapi/memberships"
|
||||
institutions: "https://www.kanopy.com/kapi/institutions/alias/{alias}"
|
||||
search: "https://www.kanopy.com/kapi/search/videos"
|
||||
videos: "https://www.kanopy.com/kapi/videos/{video_id}"
|
||||
items: "https://www.kanopy.com/kapi/videos/{video_id}/items"
|
||||
plays: "https://www.kanopy.com/kapi/plays"
|
||||
widevine: "https://www.kanopy.com/kapi/licenses/widevine/{drm_id}"
|
||||
playready: "https://www.kanopy.com/kapi/licenses/playready/{drm_id}"
|
||||
|
||||
Reference in New Issue
Block a user