Add JOYN service

AKNO/__init__.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from http.cookiejar import MozillaCookieJar
+from typing import Any, Optional
+import urllib
+import re
+import json
+import uuid
+
+import click
+from click import Context
+from bs4 import BeautifulSoup
+from pywidevine.cdm import Cdm as WidevineCdm
+
+from unshackle.core.constants import AnyTrack
+from unshackle.core.credential import Credential
+from unshackle.core.service import Service
+from unshackle.core.titles import Movies, Movie, Title_T
+from unshackle.core.tracks import Chapter, Tracks, Subtitle
+from unshackle.core.manifests.dash import DASH
+
+class AKNO(Service):
+	"""
+	Service code for alleskino (https://www.alleskino.de)
+
+	\b
+	Version: 1.0.0
+	Author: lambda
+	Authorization: Credentials
+	Robustness:
+	  Widevine:
+		L3: 1080p, AAC2.0
+	"""
+
+	GEOFENCE = ("de",)
+	TITLE_RE = r"^https?://(?:www\.)?alleskino\.de.+"
+
+	@staticmethod
+	@click.command(name="AKNO", short_help="https://alleskino.de", help=__doc__)
+	@click.argument("title", type=str)
+	@click.pass_context
+	def cli(ctx: Context, **kwargs: Any) -> AKNO:
+		return AKNO(ctx, **kwargs)
+
+	def __init__(self, ctx: Context, title: str):
+		self.title = title
+		super().__init__(ctx)
+
+	def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
+		cache = self.cache.get(f"session_{credential.sha1}")
+		if cache and not cache.expired:
+			self.auth_jwt = cache.data
+			self.log.info("Using cached JWT")
+			return
+
+		r = self.session.post(url=self.config['endpoints']['token'], data={
+			'client_id': 'filmwerte-vod-frontend',
+			'grant_type': 'password',
+			'username': credential.username,
+			'password': credential.password,
+			'scope': 'offline_access',
+			'provider': '80f88890-3034-4649-8886-1b5061d53c8b',
+		})
+		r.raise_for_status()
+		tokens = r.json()
+
+		self.auth_jwt = tokens['access_token']
+		cache.set(self.auth_jwt, expiration=tokens['expires_in'])
+		self.log.info("Logged in and acquired new JWT")
+
+	def get_titles(self) -> Title_T:
+		match = re.match(self.TITLE_RE, self.title)
+		if not match:
+			return None
+
+		r = self.session.get(self.title)
+		r.raise_for_status()
+
+		soup = BeautifulSoup(r.text, "html.parser")
+		ng_state = json.loads(soup.find("script", {"id": "ng-state"}).contents[0])
+
+		movie = ng_state['MovieModel']
+		return Movies([
+			Movie(
+				id_=movie["id"],
+				service=self.__class__,
+				name=movie["originalTitle"] or movie["title"],
+				year=movie["releaseDate"][0:4],
+				data={
+					'movie_id': movie['id'],
+					'tenant_id': ng_state['TenantModel']['id']
+				}
+			)
+		])
+
+	def get_tracks(self, title: Title_T) -> Tracks:
+		r = self.session.get(self.config['endpoints']['uris'].format(tenant_id=title.data['tenant_id'], movie_id=title.data['movie_id']), headers={
+			"Authorization": f'Bearer {self.auth_jwt}',
+		})
+		r.raise_for_status()
+		uris = r.json()
+
+		title.data['widevine_url'] = uris['widevineLicenseServerUri']
+		tracks = DASH.from_url(uris['mpegDash']).to_tracks('de')
+		return tracks
+
+	def get_chapters(self, title: Title_T) -> list[Chapter]:
+		return []
+
+	def get_widevine_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes:
+		r = self.session.post(url=title.data['widevine_url'], data=challenge, headers={
+			'Accept': '*/*',
+			'Origin': 'https://www.alleskino.de/',
+			'Referer': 'https://www.alleskino.de/',
+		})
+		r.raise_for_status()
+		return r.content

AKNO/config.yaml

@@ -0,0 +1,14 @@
+headers:
+  Accept-Language: de-DE,de;q=0.8
+  User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36
+  priority: 'u=1, i'
+  sec-ch-ua: '"Chromium";v="142", "Google Chrome";v="142", "Not_A Brand";v="99"'
+  sec-ch-ua-mobile: '?0'
+  sec-ch-ua-platform: '"Windows"'
+  sec-fetch-dest: 'empty'
+  sec-fetch-mode: 'cors'
+  sec-fetch-site: 'same-site'
+
+endpoints:
+  token: https://api.tenant.frontend.vod.filmwerte.de/connect/token
+  uris: https://api.tenant.frontend.vod.filmwerte.de/v17/{tenant_id}/movies/{movie_id}/uri

JOYN/__init__.py

@@ -0,0 +1,193 @@
+from __future__ import annotations
+
+from http.cookiejar import MozillaCookieJar
+from typing import Any, Optional
+import urllib
+import re
+import json
+import uuid
+
+import click
+from click import Context
+from bs4 import BeautifulSoup
+from pywidevine.cdm import Cdm as WidevineCdm
+
+from unshackle.core.constants import AnyTrack
+from unshackle.core.credential import Credential
+from unshackle.core.service import Service
+from unshackle.core.titles import Series, Episode, Title_T
+from unshackle.core.tracks import Chapter, Tracks, Subtitle
+from unshackle.core.manifests.dash import DASH
+
+class JOYN(Service):
+	"""
+	Service code for Joyn (https://www.joyn.de)
+
+	\b
+	Version: 1.0.0
+	Author: lambda
+	Authorization: None
+	Robustness:
+	  Widevine:
+		L3: 1080p, AAC2.0
+	"""
+
+	GEOFENCE = ("de",)
+	TITLE_RE = r"^https?://(?:www\.)?joyn\.de.+"
+	GENERIC_EPISODE_TITLE_RE = r"^Folge [0-9]+$"
+
+	@staticmethod
+	@click.command(name="JOYN", short_help="https://joyn.de", help=__doc__)
+	@click.argument("title", type=str)
+	@click.pass_context
+	def cli(ctx: Context, **kwargs: Any) -> JOYN:
+		return JOYN(ctx, **kwargs)
+
+	def __init__(self, ctx: Context, title: str):
+		self.title = title
+		super().__init__(ctx)
+
+	def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None:
+		return
+
+		# r = self.session.get(self.config['endpoints']['oauth_initial'], allow_redirects=False)
+		# r.raise_for_status()
+		# redir_url = urllib.parse.urlparse(r.headers["Location"])
+		# redir_params = urllib.parse.parse_qs(redir_url.query)
+
+		# request_id = redir_params['requestId'][0]
+		# print(request_id)
+
+		# r = self.session.get(f'https://auth.7pass.de/public-srv/public/{request_id}')
+		# r.raise_for_status()
+		# print(r.json())
+
+		# r = self.session.get('https://auth.7pass.de/registration-setup-srv/public/list', params={'acceptlanguage': 'undefined', 'requestId': request_id})
+		# r.raise_for_status()
+		# print(r.json())
+
+		# r = self.session.post(f'https://auth.7pass.de/users-srv/user/checkexists/{request_id}', json={
+		# 	'email': credential.username,
+		# 	'requestId': request_id,
+		# })
+		# r.raise_for_status()
+		# print(r.json())
+
+
+		# r = self.session.post(self.config['endpoints']['auth_initiate'], headers={
+		# 	'Origin': 'https://signin.7pass.de'
+		# }, json={
+		# 		"request_id": request_id,
+		# 		"email": credential.username,
+		# 		"medium_id": "PASSWORD",
+		# 		"usage_type": "PASSWORDLESS_AUTHENTICATION",
+		# 		"type": "PASSWORD"
+		# })
+		# r.raise_for_status()
+		# print(r.json())
+		# exit(1)
+
+	def map_episodes(self, episodes_data, season_number):
+		for episode in episodes_data:
+			episode_title = None
+			if not re.match(self.GENERIC_EPISODE_TITLE_RE, episode['title']):
+				episode_title = episode['title']
+
+			yield Episode(
+				id_=episode['video']['id'],
+				service=self.__class__,
+				name=episode_title,
+				title=episode['series']['title'].title(),
+				season=season_number,
+				number=episode['number'],
+				language="de",
+				data={'asset_id': episode['video']['id']}
+			)
+
+	def get_titles(self) -> Title_T:
+		match = re.match(self.TITLE_RE, self.title)
+		if not match:
+			return None
+
+		r = self.session.get(self.title)
+		r.raise_for_status()
+
+		soup = BeautifulSoup(r.text, "html.parser")
+		next_data = json.loads(soup.find("script", {"id": "__NEXT_DATA__"}).contents[0])
+		page_props = next_data['props']['pageProps']
+
+		if 'page' in page_props and page_props['page']['__typename'] == 'EpisodePage':
+			episodes = [page_props['page']['episode']]
+			season_number = page_props['page']['episode']['season']['number']
+		elif 'initialData' in page_props and page_props['initialData']['page']['__typename'] == 'EpisodePage':
+			episodes = page_props['initialData']['page']['episode']['season']['episodes']
+			season_number = page_props['initialData']['page']['episode']['season']['number']
+		else:
+			print('Unknown page type')
+			return None
+
+		return Series(self.map_episodes(episodes, season_number))
+
+	def get_tracks(self, title: Title_T) -> Tracks:
+		r = self.session.post(self.config['endpoints']['entitlement'], headers={
+			"Authorization": f'Bearer {self.config["keys"]["jwt"]}',
+			"joyn-b2b-context": "UNKNOWN",
+			"joyn-client-os": "UNKNOWN",
+			"joyn-client-version": "5.1387.0",
+			"joyn-platform": "web",
+			"origin": "https://www.joyn.de"
+		}, json={
+			"content_id": title.data['asset_id'],
+			"content_type": "VOD",
+		})
+
+		r.raise_for_status()
+		entitlement_token = r.json()["entitlement_token"]
+
+		r = self.session.post(
+			url=self.config['endpoints']['playlist'].format(asset_id=title.data['asset_id']),
+			headers={
+				'Authorization': f'Bearer {entitlement_token}',
+			},
+			json = {
+				"manufacturer": "unknown",
+				"platform": "browser",
+				"maxSecurityLevel": 1,
+				"streamingFormat": "dash",
+				"model": "unknown",
+				"protectionSystem": "widevine",
+				"enableDolbyAudio": False,
+				"enableSubtitles": True,
+				"variantName": "default"
+			}
+		)
+
+		r.raise_for_status()
+		playlist_data = r.json()
+
+		if playlist_data["streamingFormat"] != "dash":
+			print("Unknown streamingFormat:", playlist_data["streamingFormat"])
+
+		title.data['playlist_data'] = playlist_data
+		tracks = DASH.from_url(playlist_data["manifestUrl"], session=self.session).to_tracks('de')
+		return tracks
+
+	def get_chapters(self, title: Title_T) -> list[Chapter]:
+		return []
+
+	def get_widevine_service_certificate(self, title: Title_T, **_: Any) -> bytes:
+		r = self.session.get(url=title.data['playlist_data']['certificateUrl'], headers={
+			'Accept': '*/*',
+			'origin': 'https://www.joyn.de',
+		})
+
+		r.raise_for_status()
+		return r.content
+
+	def get_widevine_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes:
+		r = self.session.post(url=title.data['playlist_data']['licenseUrl'], data=challenge, headers={
+			'Accept': '*/*',
+			'origin': 'https://www.joyn.de',
+		})
+		r.raise_for_status()
+		return r.content

JOYN/config.yaml

@@ -0,0 +1,22 @@
+headers:
+  Accept-Language: de-DE,de;q=0.8
+  User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/144.0.0.0 Safari/537.36
+  priority: 'u=1, i'
+  sec-ch-ua: '"Chromium";v="142", "Google Chrome";v="142", "Not_A Brand";v="99"'
+  sec-ch-ua-mobile: '?0'
+  sec-ch-ua-platform: '"Windows"'
+  sec-fetch-dest: 'empty'
+  sec-fetch-mode: 'cors'
+  sec-fetch-site: 'same-site'
+
+endpoints:
+  oauth_initial: https://auth.7pass.de/authz-srv/authz?response_type=code&scope=openid+email+profile+offline_access&view_type=login&cd1=13549ef3-bef6-4b42-919a-5b1226e1c929&client_id=655e06a5-829b-40c7-8084-077b87d26f8c&prompt=consent&response_mode=query&cd2=UPSELL&cmpUcId=ec78cf80171d6f1472f357ea40b60e0812086831ff444cd8f36fb54678e44efa&cmpUcInstance=GXjugz_sV&redirect_uri=https%3A%2F%2Fwww.joyn.de%2Foauth&state=&cd9=&cd10=https%3A%2F%2Fwww.joyn.de&code_challenge=G5jpf7JDnd_Vpjc7ntLXjEaEag36wmxfazbhnE0YiHU&code_challenge_method=S256
+  auth_initiate: https://auth.7pass.de/verification-srv/v2/authenticate/initiate/PASSWORD
+  auth_authenticate: https://auth.7pass.de/verification-srv/v2/authenticate/authenticate/PASSWORD
+  auth_login: https://auth.7pass.de/login-srv/verification/login
+  entitlement: https://entitlement.p7s1.io/api/user/entitlement-token
+  graphql: https://api.joyn.de/graphql
+  playlist: https://api.vod-prd.s.joyn.de/v1/asset/{asset_id}/playlist
+
+keys:
+  jwt: replaceme

NBLA/__init__.py

@@ -4,6 +4,7 @@ from http.cookiejar import MozillaCookieJar
 from typing import Any, Optional, Union
 from functools import partial
 from pathlib import Path
+import html
 import sys
 import re
 
@@ -77,7 +78,7 @@ class NebulaSubtitle(Subtitle):
 					tag.wrap(soup.new_tag("b"))
 
 
-			text = str(soup)
+			text = html.unescape(str(soup))
 			new_subs.append(f"{count}")
 			new_subs.append(f"{caption.start} --> {caption.end}")
 			new_subs.append(f"{text}\n")