118 lines
3.3 KiB
Python
118 lines
3.3 KiB
Python
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
|