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

Added JOYNDE service

Browse Source
This commit is contained in:
blackicedbear
2026-01-24 19:26:11 +01:00
parent 0a49e79cf1
commit dae8c5b081
4 changed files with 468 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
JOYNAT/__init__.py
View File

@@ -28,7 +28,7 @@ class JOYNAT(Service):
# List of regions of which the service offers support for.
GEOFENCE = ("at")
TITLE_RE = r"^https?:\/\/www\.joyn\.at\/(?:play\/)?(?P<type>filme|serien)\/(?P<content_id>.+)$"
TITLE_RE = r"^https?:\/\/www\.joyn\.at\/(?:play\/)?(?P<type>filme|serien|compilation)\/(?P<content_id>.+)$"
AUTH_CODE_REGEX = r"[&?]code=([^&]+)"
@staticmethod
@@ -85,13 +85,11 @@ class JOYNAT(Service):
auth_url = self.config["endpoints"]["auth_url"]
token_url = self.config["endpoints"]["token_url"]
redirect_uri = self.config["endpoints"]["redirect_uri"]
client_id = self.config["client"]["idc"]
client_id = self.config["client"]["id"]
code_verifier = self.generate_code_verifier()
code_challenge = self.generate_code_challenge(code_verifier)
self.session.cookies.update(cookies)
redirect_url_request = self.session.get(
auth_url,
headers={
@@ -150,9 +148,8 @@ class JOYNAT(Service):
def _authenticate_anonymous(self) -> None:
token_url = self.config["endpoints"]["anon_auth_url"]
client_id = self.config["client"]["id"]
anon_device_id = self.config["client"]["anon_device_id"]
client_name = self.config["client"]["name"]
anon_device_id = self.generate_code_verifier()
response = self.session.post(
token_url,
@@ -160,12 +157,11 @@ class JOYNAT(Service):
**self.session.headers,
'Content-Type': 'application/json'
},
json={'client_id': client_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()
self.log.info(f"Anonymous auth response: {auth_response}")
if 'access_token' in auth_response:
self._joyn_auth_jwt = auth_response['access_token']
@@ -367,12 +363,14 @@ class JOYNAT(Service):
playlist_data = playlist.json()
if not 'manifestUrl' in playlist_data or 'licenseUrl' not in playlist_data:
self.log.error(f"Failed to fetch tracks playlist: 'manifestUrl' or 'licenseUrl' not in entitlement_data")
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']
license_url = playlist_data['licenseUrl']
# 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")

4
JOYNAT/config.yaml
View File

@@ -12,7 +12,7 @@ endpoints:
playout: "https://api.vod-prd.s.joyn.de/v1/asset/{content_id}/playlist"
client:
id: "bb4f9c4c-82ca-486d-8eb5-8aaf772df93c"
idc: "ae892ce5-8920-4f38-b272-af7d1e242579"
id: "ae892ce5-8920-4f38-b272-af7d1e242579"
anon_device_id: "bb4f9c4c-82ca-486d-8eb5-8aaf772df93c"
name: "web"
api_key: "4f0fd9f18abbe3cf0e87fdb556bc39c8"

439
JOYNDE/__init__.py Normal file
View File

@@ -0,0 +1,439 @@
from __future__ import annotations
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
import re
import json
import base64
import hashlib
import secrets
class JOYNDE(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 = ("de")
TITLE_RE = r"^https?:\/\/www\.joyn\.de\/(?:play\/)?(?P<type>filme|serien|compilation)\/(?P<content_id>.+)$"
AUTH_CODE_REGEX = r"[&?]code=([^&]+)"
@staticmethod
@click.command(name="JOYN", short_help="https://joyn.de", help=__doc__)
@click.argument("title", type=str)
@click.pass_context
def cli(ctx: click.Context, **kwargs: Any) -> JOYNDE:
return JOYNDE(ctx, **kwargs)
def __init__(self, ctx: click.Context, title: str):
self.title = title
super().__init__(ctx)
def get_session(self) -> requests.Session:
# modify the creation of the requests session (stored as self.session)
# make a super() call to take the original result and further modify it,
# or don't to make a completely fresh one if required.
session = super().get_session()
# Set default headers as specified
session.headers.update({
'Accept': '*/*',
'Accept-Language': 'de,de-DE;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'Origin': 'https://www.joyn.de/',
'Referer': 'https://www.joyn.de/',
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36 Edg/137.0.0.0',
'Joyn-Platform': 'web',
'Joyn-Distribution-Tenant': 'JOYN_DE',
'Joyn-Country': 'DE',
'Joyn-Client-Version': '5.1370.1',
})
return session
def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential) # important
if not cookies:
self._authenticate_anonymous()
else:
self._authenticate_with_cookies(cookies)
def generate_code_verifier(self):
return secrets.token_urlsafe(64)
def generate_code_challenge(self, verifier):
sha256_hash = hashlib.sha256(verifier.encode()).digest()
return base64.urlsafe_b64encode(sha256_hash).decode().rstrip("=")
def _authenticate_with_cookies(self, cookies: MozillaCookieJar) -> None:
auth_url = self.config["endpoints"]["auth_url"]
token_url = self.config["endpoints"]["token_url"]
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)
redirect_url_request = self.session.get(
auth_url,
headers={
**self.session.headers,
},
params={
'response_type': 'code',
'scope': 'openid email profile offline_access',
'view_type': 'login',
'client_id': client_id,
'prompt': 'consent',
'response_mode': 'query',
'cmpUcId': '9464e7a80af12c8cbdfbf2117f07f410b65af6af04ff3eee58ea2754590dfc83',
'redirect_uri': redirect_uri,
'code_challenge': code_challenge,
'code_challenge_method': 'S256',
},
)
redirect_url_request.raise_for_status()
redirect_url = redirect_url_request.url
# 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.")
response = self.session.post(
token_url,
headers={
**self.session.headers,
'Content-Type': 'application/json'
},
cookies=cookies,
json={
'code': auth_code,
'client_id': client_id,
'redirect_uri': redirect_uri,
'code_verifier': code_verifier
},
)
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.")
raise EnvironmentError("Cookie authentication failed: no access token in response.")
def _authenticate_anonymous(self) -> None:
token_url = self.config["endpoints"]["anon_auth_url"]
anon_device_id = self.config["client"]["anon_device_id"]
client_name = self.config["client"]["name"]
response = self.session.post(
token_url,
headers={
**self.session.headers,
'Content-Type': 'application/json'
},
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.")
raise EnvironmentError("Anonymous authentication failed: no access token in response.")
# Required 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"]
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?")
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.")
if kind == "filme":
path = f'/filme/{content_id}'
response_data = self._execute_graphql_query('PageMovieDetailNewStatic', {'path': path}, '7b49493138f2162be230fd0e3fbf5722b1db6700a8842109ed3d98979898707a')
if 'page' in response_data and 'movie' in response_data['page']:
movie_data = response_data['page']['movie']
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.")
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.")
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}'
if len(content_id.split("/")) == 1:
response_data = self._execute_graphql_query('SeriesDetailPageStatic', {'path': path}, '43cad327eeae12e14dfb629d662ebc947d78b71ec91d972ea1ef46ccdb29eede')
if 'page' in response_data and 'series' in response_data['page']:
series_data = response_data['page']['series']
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.")
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')
if 'season' in response_data:
season_data = response_data['season']
if 'episodes' not in season_data:
self.log.error("Invalid season_data data received.")
raise ValueError("Invalid season_data data received from Joyn service.")
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.")
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.")
episodes.append(
Episode(
id_=episode['id'],
service=self.__class__,
title=series_data['title'],
season=season['number'],
number=episode['number'],
name=episode['title'],
data=episode,
)
)
return Series(episodes)
elif len(content_id.split("/")) == 2:
response_data = self._execute_graphql_query('EpisodeDetailPageStatic', {'path': path}, 'c4bcacee94d38133e87879dad8d69bd8a74c7326262a1848cceb964b871c1551')
if 'page' in response_data and 'episode' in response_data['page']:
episode_data = response_data['page']['episode']
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.")
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.")
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.")
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.")
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.")
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.")
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.")
entitlement_token = entitlement_data['entitlement_token']
playlist = self.session.post(
playout_url.format(content_id=content_id),
headers={
**self.session.headers,
'Authorization': f'Bearer {entitlement_token}'
},
json={
'manufacturer': 'unknown',
'platform': 'browser',
'maxSecurityLevel': 1,
'streamingFormat': 'dash',
'model': 'unknown',
'protectionSystem': 'widevine',
'enableDolbyAudio': False,
'enableSubtitles': True,
'maxResolution': 1080,
'variantName': 'default',
}
)
playlist.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)
def get_chapters(self, title: Union[Movies, Series]) -> list[Chapter]:
# technically optional, but you must define and at least `return []`.
return Chapters()
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")
if not license_url:
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
},
data=challenge
)
if response.status_code != 200:
self.log.error(f"Failed to fetch license: {response.status_code} - {response.text}")
raise ConnectionError(response.text)
self.log.info("Successfully fetched Widevine license from Joyn service.")
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={
'operationName': operation_name,
'variables': json.dumps(variables).encode(),
'extensions': json.dumps({
'persistedQuery': {'version': 1, 'sha256Hash': persisted_query_hash}
}).encode()
}
)
response.raise_for_status()
response_data = response.json()
if 'data' not in response_data:
self.log.error(f"GraphQL response for '{operation_name}' missing 'data' field.")
raise ValueError(f"Invalid GraphQL response for '{operation_name}'.")
return response_data['data']

18
JOYNDE/config.yaml Normal file
View File

@@ -0,0 +1,18 @@
# This config file is automatically loaded into `self.config` class instance variable.
# I recommend storing information like any de-obfuscated keys, base hosts, endpoints,
# or other such configuration data.
endpoints:
anon_auth_url : "https://auth.joyn.de/auth/anonymous"
auth_url: "https://auth.7pass.de/authz-srv/authz"
token_url: "https://auth.joyn.de/auth/7pass/token"
redirect_uri: "https://www.joyn.de/oauth"
graphql_url: "https://api.joyn.de/graphql"
entitlement: "https://entitlement.p7s1.io/api/user/entitlement-token"
playout: "https://api.vod-prd.s.joyn.de/v1/asset/{content_id}/playlist"
client:
id: "655e06a5-829b-40c7-8084-077b87d26f8c"
anon_device_id: "dfd6d99e-1766-4134-932f-b1adcce5b764"
name: "web"
api_key: "4f0fd9f18abbe3cf0e87fdb556bc39c8"