thegame_1980/unshackle-services
1
0
Fork 0
mirror of https://github.com/n0stal6ic/unshackle-services.git synced 2026-07-15 17:40:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add files via upload

Browse Source
This commit is contained in:
n0stal6ic
2026-04-13 13:31:00 -05:00
committed by GitHub
parent 12f265aaa8
commit 54a7a4e743
2 changed files with 400 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
KNPY/__init__.py
+390 -181
View File
@@ -10,7 +10,7 @@ import jwt
from langcodes import Language
from unshackle.core.constants import AnyTrack
from unshackle.core.credential import Credential
from unshackle.core.manifests import DASH, HLS
from unshackle.core.manifests import DASH
from unshackle.core.search_result import SearchResult
from unshackle.core.service import Service
from unshackle.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
@@ -19,10 +19,12 @@ from unshackle.core.tracks import Subtitle, Tracks
class KNPY(Service):
"""
Service code for Kanopy (kanopy.com).
Authorization: Credentials
Service code for Kanopy (https://kanopy.com).
Author: FairTrade, n0stal6ic
Authorization: Cookies, Credentials
Security: FHD@L3
Geofence: US, CA, UK, AU, NZ
"""
TITLE_RE = r"^https?://(?:www\.)?kanopy\.com/.+/(?P<id>\d+)$"
@@ -40,122 +42,197 @@ class KNPY(Service):
super().__init__(ctx)
if not self.config:
raise ValueError("KNPY configuration not found.")
self.cdm = ctx.obj.cdm
match = re.match(self.TITLE_RE, title)
if match:
self.content_id = match.group("id")
self.search_query = None
else:
self.content_id = None
self.search_query = title
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
"user-agent": self.USER_AGENT,
})
subdomain_match = re.search(r'kanopy\.com/[a-z]{2}/([^/]+)', title)
self._subdomain = subdomain_match.group(1) if subdomain_match else None
try:
from pyplayready.cdm import Cdm as PlayReadyCdm
self.use_playready: bool = isinstance(ctx.obj.cdm, PlayReadyCdm)
except ImportError:
self.use_playready = False
self._jwt = None
self._visitor_id = None
self._user_id = None
self._domain_id = None
self.widevine_license_url = None
self.playready_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 email and password authentication required.")
if cookies:
jwt_token = None
cookie_visitor_id = None
cookie_uid = None
cache = self.cache.get("auth_token")
if cache and not cache.expired:
cached_data = cache.data
valid_token = None
for cookie in cookies:
if cookie.name == "kapi_token":
jwt_token = cookie.value
elif cookie.name == "visitor_id":
cookie_visitor_id = cookie.value
elif cookie.name == "uid":
cookie_uid = cookie.value
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.")
if valid_token:
self._jwt = valid_token
if jwt_token:
self.log.info("Attempting cookie-based authentication.")
self._jwt = jwt_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.")
try:
decoded_jwt = jwt.decode(self._jwt, options={"verify_signature": False})
exp_timestamp = decoded_jwt.get("exp")
if exp_timestamp and exp_timestamp < datetime.now(timezone.utc).timestamp():
self.log.warning("Cookie token has expired.")
if credential:
self.log.info("Falling back to credential-based authentication.")
else:
raise ValueError("Cookie token expired and no credentials provided.")
else:
jwt_data = decoded_jwt.get("data", {})
identity_id = jwt_data.get("identity_id")
uid = jwt_data.get("uid")
self._user_id = (identity_id if identity_id and str(identity_id) != "0" else None) \
or (uid if uid and str(uid) != "0" else None) \
or cookie_uid
self._visitor_id = jwt_data.get("visitor_id") or cookie_visitor_id
self.log.info(f"Successfully authenticated via cookies (user_id: {self._user_id or 0})")
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"]
return
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)
except jwt.DecodeError as e:
self.log.error(f"Failed to decode cookie token: {e}")
if credential:
self.log.info("Falling back to credential-based authentication.")
else:
raise ValueError(f"Invalid kapi_token cookie: {e}")
except KeyError as e:
self.log.error(f"Missing expected field in cookie token: {e}")
if credential:
self.log.info("Falling back to credential-based authentication.")
else:
raise ValueError(f"Invalid kapi_token structure: {e}")
else:
self.log.warning("JWT has no 'exp' claim.")
cache.set(data=cache_payload, expiration=3600)
except Exception as e:
self.log.error(f"Failed to decode JWT for caching: {e}.")
cache.set(
data={"token": self._jwt, "username": credential.username},
expiration=3600
self.log.info("No kapi_token found in cookies.")
if not credential:
raise ValueError("No kapi_token cookie found and no credentials provided.")
self.log.info("Falling back to credential-based authentication.")
if not self._jwt:
if not credential or not credential.username or not credential.password:
raise ValueError("Kanopy requires either cookies (with kapi_token) or email/password for authentication.")
cache = self.cache.get("auth_token")
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}'.")
elif isinstance(cached_data, str):
self.log.info("Found legacy cached token format.")
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("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}.")
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.")
if not self._user_id or str(self._user_id) == "0":
if not self._subdomain:
raise ValueError(
"Cannot determine library domain."
)
self.log.info(f"Looking up institution by subdomain: {self._subdomain}")
r = self.session.get(self.config["endpoints"]["institutions"].format(subdomain=self._subdomain))
r.raise_for_status()
inst = r.json()
self._domain_id = str(inst["domainId"])
self.log.info(f"Found library: {inst.get('sitename', self._subdomain)} (domain ID: {self._domain_id})")
return
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()
@@ -163,95 +240,170 @@ class KNPY(Service):
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})")
self.log.info(f"Using default library: {membership.get('sitename', 'Unknown')} (ID: {self._domain_id})")
return
for membership in memberships.get("list", []):
if membership.get("status") == "active":
self._domain_id = str(membership["domainId"])
self.log.warning(f"No default library found. Using first active domain: {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.warning(f"No active library found. Using first available domain: {self._domain_id}")
else:
raise ValueError("No active library memberships found for this user.")
raise ValueError("No library memberships found for this user.")
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.")
raise ValueError("A content ID is required to get titles.")
if not self._domain_id:
raise ValueError("Domain ID not set. Authentication failed.")
raise ValueError("Domain ID not set.")
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):
def parse_lang(taxonomies_data: dict) -> Language:
try:
langs = data.get("languages", [])
if langs and isinstance(langs, list) and len(langs) > 0:
return Language.find(langs[0])
except:
langs = taxonomies_data.get("languages", [])
if langs:
lang_name = langs[0].get("name")
if lang_name:
return Language.find(lang_name)
except (IndexError, AttributeError, TypeError):
pass
return Language.get("en")
if content_type == "video":
video_data = content_data["video"]
movie = Movie(
return Movies([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),
language=parse_lang(video_data.get("taxonomies", {})),
data=video_data,
)
return Movies([movie])
)])
elif content_type == "playlist":
playlist_data = content_data["playlist"]
playlist_data = content_data.get("playlist")
if not playlist_data:
raise ValueError("Could not find 'playlist' data dictionary.")
series_title = playlist_data["title"]
series_year = playlist_data.get("productionYear")
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
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()
r_items = self.session.get(self.config["endpoints"]["video_items"].format(video_id=self.content_id, domain_id=self._domain_id))
r_items.raise_for_status()
items_data = r_items.json()
episodes = []
for i, item in enumerate(items_data.get("list", [])):
if item.get("type") != "video":
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)
ep_match = re.search(r'Ep(?:isode)?\.?\s*(\d+)', video_data.get("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.get("taxonomies", {})),
data=video_data,
))
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,
)
)
series = Series(episodes)
series.name = series_title
series.description = playlist_data.get("descriptionHtml", "")
series.year = series_year
return series
elif content_type == "collection":
collection_data = content_data.get("collection")
if not collection_data:
raise ValueError("Could not find 'collection' data dictionary.")
series_title_main = collection_data["title"]
series_description_main = collection_data.get("descriptionHtml", "")
series_year_main = collection_data.get("productionYear")
r_seasons = self.session.get(self.config["endpoints"]["video_items"].format(video_id=self.content_id, domain_id=self._domain_id))
r_seasons.raise_for_status()
seasons_data = r_seasons.json()
all_episodes = []
self.log.info(f"Processing collection '{series_title_main}', found {len(seasons_data.get('list', []))} seasons.")
season_counter = 1
for season_item in seasons_data.get("list", []):
if season_item.get("type") != "playlist":
self.log.warning(f"Skipping unexpected item of type '{season_item.get('type')}' in collection.")
continue
season_playlist_data = season_item["playlist"]
season_id = season_playlist_data["videoId"]
season_title = season_playlist_data["title"]
self.log.info(f"Fetching episodes for season: {season_title}")
season_match = re.search(r'(?:Season|S)\s*(\d+)', season_title, re.IGNORECASE)
if season_match:
season_num = int(season_match.group(1))
else:
self.log.warning(f"Could not parse season number from '{season_title}'. Using sequential number {season_counter}.")
season_num = season_counter
season_counter += 1
r_episodes = self.session.get(self.config["endpoints"]["video_items"].format(video_id=season_id, domain_id=self._domain_id))
r_episodes.raise_for_status()
episodes_data = r_episodes.json()
for i, episode_item in enumerate(episodes_data.get("list", [])):
if episode_item.get("type") != "video":
continue
video_data = episode_item["video"]
ep_num = i + 1
ep_match = re.search(r'Ep(?:isode)?\.?\s*(\d+)', video_data.get("title", ""), re.IGNORECASE)
if ep_match:
ep_num = int(ep_match.group(1))
all_episodes.append(Episode(
id_=str(video_data["videoId"]),
service=self.__class__,
title=series_title_main,
season=season_num,
number=ep_num,
name=video_data["title"],
description=video_data.get("descriptionHtml", ""),
year=video_data.get("productionYear", series_year_main),
language=parse_lang(video_data.get("taxonomies", {})),
data=video_data,
))
if not all_episodes:
self.log.error(f"Collection '{series_title_main}' did not show any episodes.")
return Series([])
series = Series(all_episodes)
series.name = series_title_main
series.description = series_description_main
series.year = series_year_main
return series
else:
raise ValueError(f"Unsupported content type: {content_type}")
@@ -259,8 +411,7 @@ class KNPY(Service):
play_payload = {
"videoId": int(title.id),
"domainId": int(self._domain_id),
"userId": int(self._user_id),
"visitorId": self._visitor_id
"visitorId": self._visitor_id,
}
self.session.headers.setdefault("authorization", f"Bearer {self._jwt}")
@@ -281,21 +432,22 @@ class KNPY(Service):
"Playback blocked by region restriction."
)
else:
self.log.error(f"Access forbidden (HTTP 403). Response: {response_json}")
self.log.error(f"Access forbidden. Response: {response_json}")
raise PermissionError("Kanopy denied access to this video.")
r.raise_for_status()
play_data = response_json or r.json()
manifest_url = None
manifest_type = None
drm_info = {}
for manifest in play_data.get("manifests", []):
manifest_type_raw = manifest["manifestType"]
url = manifest["url"].strip()
if url.startswith("/"):
url = f"https://kanopy.com{url}"
url = f"https://www.kanopy.com{url}"
drm_type = manifest.get("drmType")
@@ -303,93 +455,152 @@ class KNPY(Service):
manifest_url = url
manifest_type = "dash"
if drm_type == "kanopyDrm":
play_id = play_data.get("playId")
if drm_type in ("kanopyDrm", "studioDrm"):
license_id = manifest.get("drmLicenseID") or f"{play_data.get('playId')}-0"
self.widevine_license_url = self.config["endpoints"]["widevine_license"].format(
license_id=f"{play_id}-0"
license_id=license_id
)
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(
self.playready_license_url = self.config["endpoints"]["playready_license"].format(
license_id=license_id
)
else:
self.log.warning(f"Unknown DASH drmType: {drm_type}")
self.widevine_license_url = None
self.playready_license_url = None
break
elif manifest_type_raw == "hls" and not manifest_url:
manifest_url = url
manifest_type = "hls"
if drm_type == "fairplay":
self.log.warning("HLS manifest uses FairPlay DRM. Skipping.")
else:
manifest_url = url
manifest_type = "hls"
self.log.warning("HLS with FairPlay DRM is not supported.")
self.widevine_license_url = None
self.log.info(f"HLS manifest selected (drmType={drm_type!r}).")
drm_info["fairplay"] = True
else:
self.widevine_license_url = None
drm_info["clear"] = True
if not manifest_url:
raise ValueError("No supported manifest found for this title.")
if manifest_type == "dash" and not self.widevine_license_url:
raise ValueError("Could not construct Widevine license URL for DASH manifest.")
raise ValueError("Could not find a DASH or HLS manifest for this title.")
if manifest_type == "dash" and not self.widevine_license_url and not self.playready_license_url:
raise ValueError("Could not construct a license URL for DASH manifest.")
self.log.info(f"Fetching {manifest_type.upper()} manifest from: {manifest_url}")
r = self.session.get(manifest_url)
r.raise_for_status()
self.session.headers.update({
"User-Agent": self.WIDEVINE_UA,
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
})
if manifest_type == "dash":
tracks = DASH.from_text(r.text, url=manifest_url).to_tracks(language=title.language)
else: # hls
tracks = HLS.from_text(r.text, url=manifest_url).to_tracks(language=title.language)
self.log.info("Successfully parsed HLS manifest.")
if not self.use_playready:
import xml.etree.ElementTree as ET
ET.register_namespace('', 'urn:mpeg:dash:schema:mpd:2011')
ET.register_namespace('cenc', 'urn:mpeg:cenc:2013')
ET.register_namespace('mspr', 'urn:microsoft:playready')
root = ET.fromstring(r.text)
for adaptation_set in root.findall('.//{urn:mpeg:dash:schema:mpd:2011}AdaptationSet'):
for cp in list(adaptation_set.findall('{urn:mpeg:dash:schema:mpd:2011}ContentProtection')):
if '9a04f079-9840-4286-ab92-e65be0885f95' in cp.get('schemeIdUri', ''):
adaptation_set.remove(cp)
mpd_text = ET.tostring(root, encoding='unicode')
else:
mpd_text = r.text
tracks = DASH.from_text(mpd_text, url=manifest_url).to_tracks(language=title.language)
elif manifest_type == "hls":
try:
from unshackle.core.manifests import HLS
tracks = HLS.from_text(r.text, url=manifest_url).to_tracks(language=title.language)
self.log.info("Successfully parsed HLS manifest")
except ImportError:
self.log.error(
"HLS manifest parser not available in unshackle. "
"Ensure your unshackle installation supports HLS."
)
raise
except Exception as e:
self.log.error(f"Failed to parse HLS manifest: {e}")
raise
else:
raise ValueError(f"Unsupported manifest type: {manifest_type}")
self.session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Accept": "*/*",
"Accept-Language": "en-US,en;q=0.9",
"Origin": "https://www.kanopy.com",
"Referer": "https://www.kanopy.com/",
})
self.session.headers.pop("x-version", None)
self.session.headers.pop("authorization", None)
for caption_data in play_data.get("captions", []):
lang = caption_data.get("language", "en")
label = caption_data.get("label", lang)
slug = label.lower()
slug = re.sub(r'[\s\[\]\(\)]+', '-', slug)
slug = re.sub(r'[^a-z0-9-]', '', slug)
slug = slug.strip('-')
track_id = f"caption-{lang}-{slug}"
for file_info in caption_data.get("files", []):
if file_info.get("type") == "webvtt":
tracks.add(Subtitle(
id_=f"caption-{lang}",
id_=track_id,
name=label,
url=file_info["url"].strip(),
codec=Subtitle.Codec.WebVTT,
language=Language.get(lang)
language=Language.get(lang),
))
break
return tracks
return tracks
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.")
license_headers = {
"Content-Type": "application/octet-stream",
"User-Agent": self.WIDEVINE_UA,
"Authorization": f"Bearer {self._jwt}",
"X-Version": self.API_VERSION
}
raise ValueError("Widevine license URL was not set.")
r = self.session.post(
self.widevine_license_url,
data=challenge,
headers=license_headers
headers={
"Content-Type": "application/octet-stream",
"User-Agent": self.WIDEVINE_UA,
"Authorization": f"Bearer {self._jwt}",
"X-Version": self.API_VERSION,
},
)
r.raise_for_status()
return r.content
def get_playready_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes:
if not self.playready_license_url:
raise ValueError("PlayReady license URL was not set.")
self.log.info(f"Requesting PlayReady license from: {self.playready_license_url}")
r = self.session.post(
self.playready_license_url,
data=challenge,
headers={
"Content-Type": "text/xml; charset=utf-8",
"User-Agent": self.WIDEVINE_UA,
"Authorization": f"Bearer {self._jwt}",
"X-Version": self.API_VERSION,
},
)
self.log.info(f"PlayReady license response: HTTP {r.status_code}")
if not r.ok:
self.log.error(f"PlayReady license error body: {r.text[:500]}")
r.raise_for_status()
return r.content
def search(self) -> Generator[SearchResult, None, None]:
if not self.search_query:
self.log.error("Search query not set. Cannot search.")
if not hasattr(self, 'search_query') or not self.search_query:
self.log.error("Search query not set.")
return
self.log.info(f"Searching for '{self.search_query}'...")
if not self._domain_id:
self._fetch_user_details()
@@ -399,7 +610,7 @@ class KNPY(Service):
"domainId": self._domain_id,
"isKids": "false",
"page": 0,
"perPage": 40
"perPage": 40,
}
r = self.session.get(self.config["endpoints"]["search"], params=params)
@@ -407,7 +618,7 @@ class KNPY(Service):
search_data = r.json()
results_list = search_data.get("list", [])
if not results_list:
self.log.warning(f"No results found for '{self.search_query}'")
return
@@ -416,14 +627,12 @@ class KNPY(Service):
video_id = item.get("videoId")
if not video_id:
continue
title = item.get("title", "Unknown Title")
yield SearchResult(
id_=str(video_id),
title=title,
label="VIDEO/SERIES",
url=f"https://www.kanopy.com/video/{video_id}"
url=f"https://www.kanopy.com/video/{video_id}",
)
def get_chapters(self, title: Title_T) -> list:
KNPY/config.yaml
+10 -9
View File
@@ -4,12 +4,13 @@ client:
widevine_ua: "KanopyApplication/6.21.0 (Linux;Android 15) AndroidXMedia3/1.8.0"
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}"
handshake: "https://www.kanopy.com/kapi/handshake"
login: "https://www.kanopy.com/kapi/login"
memberships: "https://www.kanopy.com/kapi/memberships?userId={user_id}"
institutions: "https://www.kanopy.com/kapi/institutions/alias/{subdomain}"
video_info: "https://www.kanopy.com/kapi/videos/{video_id}?domainId={domain_id}"
video_items: "https://www.kanopy.com/kapi/videos/{video_id}/items?domainId={domain_id}"
search: "https://www.kanopy.com/kapi/search/videos"
plays: "https://www.kanopy.com/kapi/plays"
widevine_license: "https://www.kanopy.com/kapi/licenses/widevine/{license_id}"
playready_license: "https://www.kanopy.com/kapi/licenses/playready/{license_id}"