forked from FairTrade/unshackle-services
Added cookie support for KNPY
This commit is contained in:
parent
fa776a590a
commit
cf3bab282c
254
KNPY/__init__.py
254
KNPY/__init__.py
@ -22,9 +22,9 @@ from unshackle.core.tracks import Subtitle, Tracks
|
|||||||
class KNPY(Service):
|
class KNPY(Service):
|
||||||
"""
|
"""
|
||||||
Service code for Kanopy (kanopy.com).
|
Service code for Kanopy (kanopy.com).
|
||||||
Version: 1.0.0
|
Version: 1.1.0
|
||||||
|
|
||||||
Auth: Credential (username + password)
|
Auth: Cookies (kapi_token) or Credential (username + password)
|
||||||
Security: FHD@L3
|
Security: FHD@L3
|
||||||
|
|
||||||
Handles both Movies and Series (Playlists).
|
Handles both Movies and Series (Playlists).
|
||||||
@ -32,7 +32,6 @@ class KNPY(Service):
|
|||||||
Caching included
|
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+)$"
|
TITLE_RE = r"^https?://(?:www\.)?kanopy\.com/.+/(?P<id>\d+)$"
|
||||||
GEOFENCE = ()
|
GEOFENCE = ()
|
||||||
NO_SUBTITLES = False
|
NO_SUBTITLES = False
|
||||||
@ -74,110 +73,194 @@ class KNPY(Service):
|
|||||||
self.widevine_license_url = None
|
self.widevine_license_url = None
|
||||||
|
|
||||||
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> 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.")
|
Authenticate using either cookies or credentials.
|
||||||
|
|
||||||
cache = self.cache.get("auth_token")
|
|
||||||
|
|
||||||
if cache and not cache.expired:
|
Cookie-based auth: Requires 'kapi_token' cookie from browser.
|
||||||
cached_data = cache.data
|
Credential-based auth: Requires email and password.
|
||||||
valid_token = None
|
"""
|
||||||
|
|
||||||
if isinstance(cached_data, dict) and "token" in cached_data:
|
if cookies:
|
||||||
if cached_data.get("username") == credential.username:
|
jwt_token = None
|
||||||
valid_token = cached_data["token"]
|
cookie_visitor_id = None
|
||||||
self.log.info("Using cached authentication token")
|
cookie_uid = None
|
||||||
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):
|
# Extract relevant cookies
|
||||||
self.log.info("Found legacy cached token format. Re-authenticating to ensure correct user.")
|
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 valid_token:
|
if jwt_token:
|
||||||
self._jwt = valid_token
|
self.log.info("Attempting cookie-based authentication...")
|
||||||
|
self._jwt = jwt_token
|
||||||
self.session.headers.update({"authorization": f"Bearer {self._jwt}"})
|
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:
|
||||||
try:
|
# Decode JWT to extract user information
|
||||||
decoded_jwt = jwt.decode(self._jwt, options={"verify_signature": False})
|
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"]
|
# Check if token is expired
|
||||||
self.log.info(f"Extracted user_id and visitor_id from cached token.")
|
exp_timestamp = decoded_jwt.get("exp")
|
||||||
self._fetch_user_details()
|
if exp_timestamp and exp_timestamp < datetime.now(timezone.utc).timestamp():
|
||||||
return
|
self.log.warning("Cookie token has expired.")
|
||||||
except (KeyError, jwt.DecodeError) as e:
|
if credential:
|
||||||
self.log.error(f"Could not decode cached token: {e}. Re-authenticating.")
|
self.log.info("Falling back to credential-based authentication...")
|
||||||
|
else:
|
||||||
|
raise ValueError("Cookie token expired and no credentials provided.")
|
||||||
|
else:
|
||||||
|
# Extract user data from JWT
|
||||||
|
jwt_data = decoded_jwt.get("data", {})
|
||||||
|
self._user_id = jwt_data.get("uid") or cookie_uid
|
||||||
|
self._visitor_id = jwt_data.get("visitor_id") or cookie_visitor_id
|
||||||
|
|
||||||
self.log.info("Performing handshake to get visitor token...")
|
if not self._user_id:
|
||||||
r = self.session.get(self.config["endpoints"]["handshake"])
|
raise ValueError("Could not extract user_id from cookie token")
|
||||||
r.raise_for_status()
|
|
||||||
handshake_data = r.json()
|
self.log.info(f"Successfully authenticated via cookies (user_id: {self._user_id})")
|
||||||
self._visitor_id = handshake_data["visitorId"]
|
|
||||||
initial_jwt = handshake_data["jwt"]
|
# Fetch user library memberships to get domain_id
|
||||||
|
self._fetch_user_details()
|
||||||
self.log.info(f"Logging in as {credential.username}...")
|
return
|
||||||
login_payload = {
|
|
||||||
"credentialType": "email",
|
except jwt.DecodeError as e:
|
||||||
"emailUser": {
|
self.log.error(f"Failed to decode cookie token: {e}")
|
||||||
"email": credential.username,
|
if credential:
|
||||||
"password": credential.password
|
self.log.info("Falling back to credential-based authentication...")
|
||||||
}
|
else:
|
||||||
}
|
raise ValueError(f"Invalid kapi_token cookie: {e}")
|
||||||
r = self.session.post(
|
except KeyError as e:
|
||||||
self.config["endpoints"]["login"],
|
self.log.error(f"Missing expected field in cookie token: {e}")
|
||||||
json=login_payload,
|
if credential:
|
||||||
headers={"authorization": f"Bearer {initial_jwt}"}
|
self.log.info("Falling back to credential-based authentication...")
|
||||||
)
|
else:
|
||||||
r.raise_for_status()
|
raise ValueError(f"Invalid kapi_token structure: {e}")
|
||||||
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:
|
else:
|
||||||
self.log.warning("JWT has no 'exp' claim, caching for 1 hour as a fallback.")
|
self.log.info("No kapi_token found in cookies.")
|
||||||
cache.set(data=cache_payload, expiration=3600)
|
if not credential:
|
||||||
except Exception as e:
|
raise ValueError("No kapi_token cookie found and no credentials provided.")
|
||||||
self.log.error(f"Failed to decode JWT for caching: {e}. Caching for 1 hour as a fallback.")
|
self.log.info("Falling back to credential-based authentication...")
|
||||||
cache.set(
|
|
||||||
data={"token": self._jwt, "username": credential.username},
|
if not self._jwt: # Only proceed if not already authenticated via cookies
|
||||||
expiration=3600
|
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.")
|
||||||
|
|
||||||
|
# Check for cached credential-based token
|
||||||
|
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.")
|
||||||
|
|
||||||
|
# Perform fresh login with credentials
|
||||||
|
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()
|
||||||
|
|
||||||
|
# Cache the token
|
||||||
|
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):
|
def _fetch_user_details(self):
|
||||||
|
"""Fetch user library memberships to determine the active domain_id."""
|
||||||
self.log.info("Fetching user library memberships...")
|
self.log.info("Fetching user library memberships...")
|
||||||
r = self.session.get(self.config["endpoints"]["memberships"].format(user_id=self._user_id))
|
r = self.session.get(self.config["endpoints"]["memberships"].format(user_id=self._user_id))
|
||||||
r.raise_for_status()
|
r.raise_for_status()
|
||||||
memberships = r.json()
|
memberships = r.json()
|
||||||
|
|
||||||
|
# Look for the default active membership
|
||||||
for membership in memberships.get("list", []):
|
for membership in memberships.get("list", []):
|
||||||
if membership.get("status") == "active" and membership.get("isDefault", False):
|
if membership.get("status") == "active" and membership.get("isDefault", False):
|
||||||
self._domain_id = str(membership["domainId"])
|
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 domain: {membership.get('sitename', 'Unknown')} (ID: {self._domain_id})")
|
||||||
return
|
return
|
||||||
|
|
||||||
|
# Fallback to first active membership
|
||||||
|
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"):
|
if memberships.get("list"):
|
||||||
self._domain_id = str(memberships["list"][0]["domainId"])
|
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:
|
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:
|
def get_titles(self) -> Titles_T:
|
||||||
if not self.content_id:
|
if not self.content_id:
|
||||||
@ -456,9 +539,6 @@ class KNPY(Service):
|
|||||||
|
|
||||||
title = item.get("title", "Unknown Title")
|
title = item.get("title", "Unknown Title")
|
||||||
|
|
||||||
# Since the search API doesn't explicitly return "type",
|
|
||||||
# we provide a generic label or try to guess.
|
|
||||||
# In your get_titles logic, you handle both, so we point to the watch URL.
|
|
||||||
yield SearchResult(
|
yield SearchResult(
|
||||||
id_=str(video_id),
|
id_=str(video_id),
|
||||||
title=title,
|
title=title,
|
||||||
@ -467,4 +547,4 @@ class KNPY(Service):
|
|||||||
)
|
)
|
||||||
|
|
||||||
def get_chapters(self, title: Title_T) -> list:
|
def get_chapters(self, title: Title_T) -> list:
|
||||||
return []
|
return []
|
||||||
Loading…
x
Reference in New Issue
Block a user