diff --git a/packages/envied/src/envied/envied-example-full-reference.yaml b/packages/envied/src/envied/envied-example-full-reference.yaml index 60ca8c9..2b1aeb2 100644 --- a/packages/envied/src/envied/envied-example-full-reference.yaml +++ b/packages/envied/src/envied/envied-example-full-reference.yaml @@ -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" diff --git a/packages/envied/src/envied/envied-working-example.yaml b/packages/envied/src/envied/envied-working-example.yaml index 195ce77..4214d36 100644 --- a/packages/envied/src/envied/envied-working-example.yaml +++ b/packages/envied/src/envied/envied-working-example.yaml @@ -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/ diff --git a/packages/envied/src/envied/services/CRAV/__init__.py b/packages/envied/src/envied/services/CRAV/__init__.py new file mode 100644 index 0000000..ace144d --- /dev/null +++ b/packages/envied/src/envied/services/CRAV/__init__.py @@ -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})?/(?Pmovie|series|play)/(?:[a-z0-9-]+/)?)?([a-z0-9-]+?)(?:-(?P\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] diff --git a/packages/envied/src/envied/services/CRAV/config.yaml b/packages/envied/src/envied/services/CRAV/config.yaml new file mode 100644 index 0000000..f45eac4 --- /dev/null +++ b/packages/envied/src/envied/services/CRAV/config.yaml @@ -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=" diff --git a/packages/envied/src/envied/services/KNPY/__init__.py b/packages/envied/src/envied/services/KNPY/__init__.py index 8b1f9e7..c7cd3c3 100644 --- a/packages/envied/src/envied/services/KNPY/__init__.py +++ b/packages/envied/src/envied/services/KNPY/__init__.py @@ -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\d+)$" - GEOFENCE = () - NO_SUBTITLES = False + # GEOFENCE = () + ALIASES = ("kanopy",) + TITLE_RE = r"^(?:.*?/)?(?Pwatch/video|video|watch|product)/(?P\d+)(?:/(?P\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 - match = re.match(self.TITLE_RE, title) - if match: - self.content_id = match.group("id") - 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(self.config["headers"]) - self.session.headers.update({ - "x-version": self.API_VERSION, - "user-agent": self.USER_AGENT - }) + 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._jwt = None - self._visitor_id = None - self._user_id = None - self._domain_id = None - self.widevine_license_url = None + cache = self.cache.get(f"tokens_{credential.sha1}") - 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") - 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_service_certificate(self, *, challenge: bytes, title: Episode | Movie, track: Any) -> bytes | str | None: + return None - 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 + 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 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 + + license_url = self.config["endpoints"]["playready"].format(drm_id=drm_id) + + 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"] + } + + 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 [] - - # 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() + # Service-specific methods - # results = [] - # for item in search_data.get("list", []): - # item_type = item.get("type") - # if item_type not in ["playlist", "video"]: - # continue + 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 - # video_id = item.get("videoId") - # title = item.get("title", "No Title") - # label = "Series" if item_type == "playlist" else "Movie" - - # 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 [] diff --git a/packages/envied/src/envied/services/KNPY/config.yaml b/packages/envied/src/envied/services/KNPY/config.yaml index 7e61f6f..9c8021b 100644 --- a/packages/envied/src/envied/services/KNPY/config.yaml +++ b/packages/envied/src/envied/services/KNPY/config.yaml @@ -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}" \ No newline at end of file + 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}"