blackicedbear/devine_services
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(JOYNAT): improve code style and best practices

Browse Source
This commit is contained in:
blackicedbear
2026-03-02 00:57:21 +01:00
parent 49a14161cf
commit 5bf50ecba4
1 changed files with 227 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files

450
JOYNAT/__init__.py
View File

@@ -1,32 +1,31 @@
from __future__ import annotations
import base64
import hashlib
import json
import re
import secrets
from http.cookiejar import MozillaCookieJar
from typing import Any, Optional, Union
import click
import requests
from devine.core.titles import Episode, Movie, Movies, Series
from devine.core.manifests import DASH
from devine.core.constants import AnyTrack
from devine.core.service import Service
from devine.core.titles import Movies, Series
from devine.core.tracks import Chapter, Tracks
from devine.core.credential import Credential
from devine.core.tracks import Chapters, Tracks, Track
from devine.core.manifests import DASH
from devine.core.service import Service
from devine.core.titles import Episode, Movie, Movies, Series
from devine.core.tracks import Chapter, Chapters, Tracks
import re
import json
import base64
import hashlib
import secrets
class JOYNAT(Service):
"""Joyn Austria (joyn.at) streaming service."""
# List of Service Aliases. Do NOT include the Service Tag. All aliases must be lowercase.
ALIASES = ()
# List of regions of which the service offers support for.
GEOFENCE = ("at")
GEOFENCE = ("at",)
TITLE_RE = r"^https?:\/\/www\.joyn\.at\/(?:play\/)?(?P<type>filme|serien|compilation)\/(?P<content_id>.+)$"
AUTH_CODE_REGEX = r"[&?]code=([^&]+)"
@@ -76,10 +75,14 @@ class JOYNAT(Service):
else:
self._authenticate_with_cookies(cookies)
def generate_code_verifier(self):
@staticmethod
def _generate_code_verifier() -> str:
"""Generate a PKCE code verifier."""
return secrets.token_urlsafe(64)
def generate_code_challenge(self, verifier):
@staticmethod
def _generate_code_challenge(verifier: str) -> str:
"""Generate a PKCE code challenge from a verifier."""
sha256_hash = hashlib.sha256(verifier.encode()).digest()
return base64.urlsafe_b64encode(sha256_hash).decode().rstrip("=")
@@ -89,14 +92,11 @@ class JOYNAT(Service):
redirect_uri = self.config["endpoints"]["redirect_uri"]
client_id = self.config["client"]["id"]
code_verifier = self.generate_code_verifier()
code_challenge = self.generate_code_challenge(code_verifier)
code_verifier = self._generate_code_verifier()
code_challenge = self._generate_code_challenge(code_verifier)
redirect_url_request = self.session.get(
auth_url,
headers={
**self.session.headers,
},
params={
'response_type': 'code',
'scope': 'openid email profile offline_access',
@@ -116,17 +116,15 @@ class JOYNAT(Service):
# Find the auth_code using regex
auth_code_match = re.search(self.AUTH_CODE_REGEX, redirect_url)
if auth_code_match:
auth_code = auth_code_match.group(1)
self.log.debug(f"Auth Code: {auth_code}")
else:
self.log.error("Authorization code not found in redirect URL.")
raise EnvironmentError("Could not find authorization code.")
if not auth_code_match:
raise EnvironmentError("Authorization code not found in redirect URL.")
auth_code = auth_code_match.group(1)
self.log.debug(f"Auth Code: {auth_code}")
response = self.session.post(
token_url,
headers={
**self.session.headers,
'Content-Type': 'application/json'
},
cookies=cookies,
@@ -140,13 +138,12 @@ class JOYNAT(Service):
response.raise_for_status()
auth_response = response.json()
if 'access_token' in auth_response:
self._joyn_auth_jwt = auth_response['access_token']
self.log.info("Successfully authenticated with cookies.")
else:
self.log.error("No access_token found in response.")
if 'access_token' not in auth_response:
raise EnvironmentError("Cookie authentication failed: no access token in response.")
self._joyn_auth_jwt = auth_response['access_token']
self.log.info("Successfully authenticated with cookies.")
def _authenticate_anonymous(self) -> None:
token_url = self.config["endpoints"]["anon_auth_url"]
@@ -156,212 +153,237 @@ class JOYNAT(Service):
response = self.session.post(
token_url,
headers={
**self.session.headers,
'Content-Type': 'application/json'
'Content-Type': 'application/json',
},
json={
'client_id': anon_device_id,
'client_name': client_name,
'anon_device_id': anon_device_id,
},
json={'client_id': anon_device_id, 'client_name': client_name, 'anon_device_id': anon_device_id},
)
response.raise_for_status()
auth_response = response.json()
if 'access_token' in auth_response:
self._joyn_auth_jwt = auth_response['access_token']
self.log.info("Authenticated anonymously with Joyn service successfully.")
else:
self.log.error("No access_token found in response.")
if 'access_token' not in auth_response:
raise EnvironmentError("Anonymous authentication failed: no access token in response.")
self._joyn_auth_jwt = auth_response['access_token']
self.log.info("Authenticated anonymously with Joyn service.")
# Required methods:
def _is_age_restricted(self, data: dict, title_name: str) -> bool:
"""Check if content is age-restricted and should be skipped.
Returns True if the content has an age rating >= 16 and --age-bypass
was not specified.
"""
age_rating = data.get('ageRating') or {}
min_age = age_rating.get('minAge', 0)
if min_age >= 16 and not self.age_bypass:
self.log.warning(f"Skipping '{title_name}' due to age rating ({min_age}+). Use --age-bypass to download.")
return True
return False
@staticmethod
def _validate_required_fields(data: dict, fields: list[str], context: str) -> None:
"""Raise ValueError if any required fields are missing from data."""
missing = [f for f in fields if f not in data]
if missing:
raise ValueError(f"Missing required fields {missing} in {context}.")
def _validate_video_field(self, data: dict, context: str) -> None:
"""Validate that a 'video.id' field exists in the data."""
if 'video' not in data or 'id' not in data['video']:
raise ValueError(f"Missing 'video.id' in {context}.")
# ------------------------------------------------------------------
# Required service methods
# ------------------------------------------------------------------
def get_titles(self) -> Union[Movies, Series]:
graphql_url = self.config["endpoints"]["graphql_url"]
api_key = self.config["client"]["api_key"]
client_name = self.config["client"]["name"]
match = re.match(self.TITLE_RE, self.title)
if not match:
raise ValueError(f"Invalid title URL format: {self.title}")
try:
match = re.match(self.TITLE_RE, self.title)
if not match:
self.log.error(f"Invalid title URL format: {self.title}")
raise ValueError("Invalid title URL format.")
kind = match.group("type")
content_id = match.group("content_id")
except Exception:
raise ValueError("Could not parse ID from title - is the URL correct?")
kind = match.group("type")
content_id = match.group("content_id")
if not kind or not content_id:
self.log.error(f"Invalid title URL: {self.title}. 'kind' or 'content_id' is missing.")
raise ValueError("Invalid title URL: 'kind' or 'content_id' is missing.")
raise ValueError(f"Invalid title URL: {self.title}. 'kind' or 'content_id' is missing.")
if kind == "filme":
path = f'/filme/{content_id}'
return self._get_movie_titles(content_id)
elif kind == "serien":
return self._get_series_titles(content_id)
response_data = self._execute_graphql_query('PageMovieDetailNewStatic', {'path': path}, '7b49493138f2162be230fd0e3fbf5722b1db6700a8842109ed3d98979898707a')
raise ValueError(f"Unsupported content type: '{kind}'.")
if 'page' in response_data and 'movie' in response_data['page']:
movie_data = response_data['page']['movie']
def _get_movie_titles(self, content_id: str) -> Movies:
"""Fetch and return movie title data."""
path = f'/filme/{content_id}'
response_data = self._execute_graphql_query(
'PageMovieDetailNewStatic', {'path': path},
'7b49493138f2162be230fd0e3fbf5722b1db6700a8842109ed3d98979898707a',
)
if 'id' not in movie_data or 'title' not in movie_data or 'productionYear' not in movie_data:
self.log.error("Invalid movie_data data received.")
raise ValueError("Invalid movie_data data received from Joyn service.")
movie_data = response_data.get('page', {}).get('movie')
if not movie_data:
raise ValueError("Failed to fetch movie data from Joyn service.")
if 'video' not in movie_data or 'id' not in movie_data['video']:
self.log.error("Invalid movie_data data received.")
raise ValueError("Invalid movie_data data received from Joyn service.")
self._validate_required_fields(movie_data, ['id', 'title', 'productionYear'], 'movie_data')
self._validate_video_field(movie_data, 'movie_data')
min_age = movie_data.get('ageRating', {}).get('minAge', 0) if movie_data.get('ageRating') else 0
if min_age >= 16 and not self.age_bypass:
self.log.warning(f"Skipping '{movie_data.get('title')}' due to age rating ({min_age}+). Use --age-bypass to download.")
return Movies([])
if self._is_age_restricted(movie_data, movie_data['title']):
return Movies([])
return Movies(
[
Movie(
id_=movie_data['id'],
service=self.__class__,
name=movie_data['title'],
data=movie_data,
year=movie_data['productionYear'],
)
]
)
return Movies([
Movie(
id_=movie_data['id'],
service=self.__class__,
name=movie_data['title'],
data=movie_data,
year=movie_data['productionYear'],
)
])
if kind == "serien":
path = f'/serien/{content_id}'
def _get_series_titles(self, content_id: str) -> Series:
"""Fetch and return series/episode title data."""
path = f'/serien/{content_id}'
parts = content_id.split("/")
if len(content_id.split("/")) == 1:
response_data = self._execute_graphql_query('SeriesDetailPageStatic', {'path': path}, '43cad327eeae12e14dfb629d662ebc947d78b71ec91d972ea1ef46ccdb29eede')
if len(parts) == 1:
return self._get_full_series(path)
elif len(parts) == 2:
return self._get_single_episode(path)
if 'page' in response_data and 'series' in response_data['page']:
series_data = response_data['page']['series']
raise ValueError(f"Unexpected series URL depth: '{content_id}'.")
if 'title' not in series_data or 'allSeasons' not in series_data:
self.log.error("Invalid series_data data received.")
raise ValueError("Invalid series_data data received from Joyn service.")
def _get_full_series(self, path: str) -> Series:
"""Fetch all episodes across all seasons of a series."""
response_data = self._execute_graphql_query(
'SeriesDetailPageStatic', {'path': path},
'43cad327eeae12e14dfb629d662ebc947d78b71ec91d972ea1ef46ccdb29eede',
)
episodes = []
for season in series_data['allSeasons']:
if 'id' not in season or 'number' not in season:
self.log.error("Invalid series_data data received.")
raise ValueError("Invalid series_data data received from Joyn service.")
#
response_data = self._execute_graphql_query('Season', {"id": season["id"]}, 'ee2396bb1b7c9f800e5cefd0b341271b7213fceb4ebe18d5a30dab41d703009f')
series_data = response_data.get('page', {}).get('series')
if not series_data:
raise ValueError("Failed to fetch series data from Joyn service.")
if 'season' in response_data:
season_data = response_data['season']
self._validate_required_fields(series_data, ['title', 'allSeasons'], 'series_data')
if 'episodes' not in season_data:
self.log.error("Invalid season_data data received.")
raise ValueError("Invalid season_data data received from Joyn service.")
episodes = []
for season in series_data['allSeasons']:
self._validate_required_fields(season, ['id', 'number'], 'season')
for episode in season_data['episodes']:
if 'id' not in episode or 'title' not in episode or 'number' not in episode:
self.log.error("Invalid episode data received.")
raise ValueError("Invalid episode data received from Joyn service.")
season_response = self._execute_graphql_query(
'Season', {'id': season['id']},
'ee2396bb1b7c9f800e5cefd0b341271b7213fceb4ebe18d5a30dab41d703009f',
)
if 'video' not in episode or 'id' not in episode['video']:
self.log.error("Invalid episode data received.")
raise ValueError("Invalid episode data received from Joyn service.")
season_data = season_response.get('season')
if not season_data:
continue
min_age = episode.get('ageRating', {}).get('minAge', 0) if episode.get('ageRating') else 0
if min_age >= 16 and not self.age_bypass:
self.log.warning(f"Skipping episode '{episode.get('title')}' due to age rating ({min_age}+). Use --age-bypass to download.")
continue
self._validate_required_fields(season_data, ['episodes'], 'season_data')
episodes.append(
Episode(
id_=episode['id'],
service=self.__class__,
title=series_data['title'],
season=season['number'],
number=episode['number'],
name=episode['title'],
data=episode,
)
)
for episode in season_data['episodes']:
self._validate_required_fields(episode, ['id', 'title', 'number'], 'episode')
self._validate_video_field(episode, 'episode')
return Series(episodes)
if self._is_age_restricted(episode, episode['title']):
continue
elif len(content_id.split("/")) == 2:
response_data = self._execute_graphql_query('EpisodeDetailPageStatic', {'path': path}, 'c4bcacee94d38133e87879dad8d69bd8a74c7326262a1848cceb964b871c1551')
episodes.append(Episode(
id_=episode['id'],
service=self.__class__,
title=series_data['title'],
season=season['number'],
number=episode['number'],
name=episode['title'],
data=episode,
))
if 'page' in response_data and 'episode' in response_data['page']:
episode_data = response_data['page']['episode']
return Series(episodes)
if 'id' not in episode_data or 'title' not in episode_data or 'number' not in episode_data:
self.log.error("Invalid episode_data data received.")
raise ValueError("Invalid episode_data data received from Joyn service.")
def _get_single_episode(self, path: str) -> Series:
"""Fetch a single episode by its direct URL."""
response_data = self._execute_graphql_query(
'EpisodeDetailPageStatic', {'path': path},
'c4bcacee94d38133e87879dad8d69bd8a74c7326262a1848cceb964b871c1551',
)
if 'season' not in episode_data or 'number' not in episode_data['season']:
self.log.error("Invalid episode_data data received.")
raise ValueError("Invalid episode_data data received from Joyn service.")
episode_data = response_data.get('page', {}).get('episode')
if not episode_data:
raise ValueError("Failed to fetch episode data from Joyn service.")
if 'series' not in episode_data or 'title' not in episode_data['series']:
self.log.error("Invalid episode_data data received.")
raise ValueError("Invalid episode_data data received from Joyn service.")
self._validate_required_fields(episode_data, ['id', 'title', 'number'], 'episode_data')
self._validate_required_fields(episode_data.get('season', {}), ['number'], 'episode_data.season')
self._validate_required_fields(episode_data.get('series', {}), ['title'], 'episode_data.series')
self._validate_video_field(episode_data, 'episode_data')
if 'video' not in episode_data or 'id' not in episode_data['video']:
self.log.error("Invalid episode_data data received.")
raise ValueError("Invalid episode_data data received from Joyn service.")
if self._is_age_restricted(episode_data, episode_data['title']):
return Series([])
min_age = episode_data.get('ageRating', {}).get('minAge', 0) if episode_data.get('ageRating') else 0
if min_age >= 16 and not self.age_bypass:
self.log.warning(f"Skipping episode '{episode_data['title']}' due to age rating ({min_age}+). Use --age-bypass to download.")
return Series([])
return Series(
[
Episode(
id_=episode_data['id'],
service=self.__class__,
title=episode_data['series']['title'],
season=episode_data['season']['number'],
number=episode_data['number'],
name=episode_data['title'],
data=episode_data,
)
]
)
self.log.error(f"Failed to fetch Movie data: {response.status_code} - {response.text}")
raise EnvironmentError("Failed to fetch Movie data from Joyn service.")
return Series([
Episode(
id_=episode_data['id'],
service=self.__class__,
title=episode_data['series']['title'],
season=episode_data['season']['number'],
number=episode_data['number'],
name=episode_data['title'],
data=episode_data,
)
])
def get_tracks(self, title: Union[Episode, Movie]) -> Tracks:
entitlement_url = self.config["endpoints"]["entitlement"]
playout_url = self.config["endpoints"]["playout"]
if not isinstance(title, Episode) and not isinstance(title, Movie):
self.log.error(f"Unsupported title type: {type(title)}. Expected Series or Movies.")
raise ValueError(f"Unsupported title type: {type(title)}. Expected Series or Movies.")
if not isinstance(title, (Episode, Movie)):
raise TypeError(f"Expected Episode or Movie, got {type(title).__name__}.")
content_id = title.data['video']['id']
entitlement = self.session.post(
entitlement_url,
headers={
**self.session.headers,
'Authorization': f'Bearer {self._joyn_auth_jwt}'
},
json={'content_id': content_id, 'content_type': 'VOD'}
)
entitlement.raise_for_status()
entitlement_data = entitlement.json()
if 'entitlement_token' not in entitlement_data:
self.log.error(f"Failed to fetch tracks entitlement: 'entitlement_token' not in entitlement_data")
raise EnvironmentError("Failed to fetch tracks entitlement from Joyn service.")
# Step 1: Obtain entitlement token
entitlement_data = self._fetch_entitlement(content_id)
entitlement_token = entitlement_data['entitlement_token']
playlist = self.session.post(
playout_url.format(content_id=content_id),
# Step 2: Obtain playlist/manifest
playlist_data = self._fetch_playlist(content_id, entitlement_token)
manifest_url = playlist_data['manifestUrl']
license_url = playlist_data.get('licenseUrl')
# Step 3: Parse DASH manifest into tracks
all_tracks = DASH.from_url(manifest_url, self.session).to_tracks(language="de")
for track in all_tracks:
if track.data is None:
track.data = {}
track.data['license_url'] = license_url
return Tracks(all_tracks)
def _fetch_entitlement(self, content_id: str) -> dict:
"""Request an entitlement token for the given content."""
response = self.session.post(
self.config["endpoints"]["entitlement"],
headers={
**self.session.headers,
'Authorization': f'Bearer {entitlement_token}'
'Authorization': f'Bearer {self._joyn_auth_jwt}',
},
json={'content_id': content_id, 'content_type': 'VOD'},
)
response.raise_for_status()
data = response.json()
if 'entitlement_token' not in data:
raise EnvironmentError("Entitlement response missing 'entitlement_token'.")
return data
def _fetch_playlist(self, content_id: str, entitlement_token: str) -> dict:
"""Request the playout/manifest URL for the given content."""
response = self.session.post(
self.config["endpoints"]["playout"].format(content_id=content_id),
headers={
'Authorization': f'Bearer {entitlement_token}',
},
json={
'manufacturer': 'unknown',
@@ -374,31 +396,14 @@ class JOYNAT(Service):
'enableSubtitles': True,
'maxResolution': 1080,
'variantName': 'default',
}
},
)
playlist.raise_for_status()
response.raise_for_status()
playlist_data = playlist.json()
if not 'manifestUrl' in playlist_data:
self.log.error(f"Failed to fetch tracks playlist: 'manifestUrl' not in entitlement_data")
raise EnvironmentError("Failed to fetch tracks playlist from Joyn service.")
manifest_url = playlist_data['manifestUrl']
# Get license_url or set to None if not present
license_url = playlist_data.get('licenseUrl', None)
all_tracks = DASH.from_url(manifest_url, self.session).to_tracks(language="de")
# Attach license_url to each track's data dictionary
for tr in all_tracks:
if tr.data is None:
tr.data = {}
tr.data['license_url'] = license_url
# Return a new Tracks object containing all collected tracks
return Tracks(all_tracks)
data = response.json()
if 'manifestUrl' not in data:
raise EnvironmentError("Playlist response missing 'manifestUrl'.")
return data
def get_chapters(self, title: Union[Movies, Series]) -> list[Chapter]:
# technically optional, but you must define and at least `return []`.
@@ -407,34 +412,31 @@ class JOYNAT(Service):
def get_widevine_service_certificate(self, *, challenge: bytes, title: Union[Movies, Series], track: AnyTrack) -> Union[bytes, str]:
return None
def get_widevine_license(self, *, challenge: bytes, title: Union[Movies, Series], track: AnyTrack) -> Optional[Union[bytes, str]]:
# Safely extract license_url from track.data
if hasattr(track, "data") and track.data:
license_url = track.data.get("license_url")
def get_widevine_license(
self, *, challenge: bytes, title: Union[Movies, Series], track: AnyTrack,
) -> Optional[Union[bytes, str]]:
"""Obtain a Widevine license for the given track."""
license_url = getattr(track, 'data', {}).get('license_url') if track.data else None
if not license_url:
raise ValueError("No license_url in track.data")
raise ValueError("No license_url in track.data.")
response = self.session.post(
license_url,
headers={
**self.session.headers,
'Content-Type': 'application/octet-stream',
'x-auth-token': self._joyn_auth_jwt
'x-auth-token': self._joyn_auth_jwt,
},
data=challenge
data=challenge,
)
if response.status_code != 200:
self.log.error(f"Failed to fetch license: {response.status_code} - {response.text}")
raise ConnectionError(response.text)
response.raise_for_status()
self.log.info("Successfully fetched Widevine license from Joyn service.")
self.log.info("Successfully fetched Widevine license.")
return response.content
def _execute_graphql_query(self, operation_name: str, variables: dict, persisted_query_hash: str) -> dict:
response = self.session.get(
self.config["endpoints"]["graphql_url"],
headers={
**self.session.headers,
'X-Api-Key': self.config["client"]["api_key"]
},
params={