unshackle-services/JOYN/__init__.py

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