diff --git a/PTHS/__init__.py b/PTHS/__init__.py deleted file mode 100644 index 33df9e3..0000000 --- a/PTHS/__init__.py +++ /dev/null @@ -1,149 +0,0 @@ -import json -import re -from typing import Optional -from http.cookiejar import CookieJar -from langcodes import Language -import click - -from unshackle.core.constants import AnyTrack -from unshackle.core.credential import Credential -from unshackle.core.manifests import DASH -from unshackle.core.service import Service -from unshackle.core.titles import Movie, Movies, Title_T, Titles_T -from unshackle.core.tracks import Tracks - - -class PTHS(Service): - """ - Service code for Pathé Thuis (pathe-thuis.nl) - Version: 1.0.0 - - Security: SD @ L3 (Widevine) - FHD @ L1 - Authorization: Cookies or authentication token - - Supported: - • Movies → https://www.pathe-thuis.nl/film/{id} - - Note: - Pathé Thuis does not have episodic content, only movies. - """ - - TITLE_RE = ( - r"^(?:https?://(?:www\.)?pathe-thuis\.nl/film/)?(?P\d+)(?:/[^/]+)?$" - ) - GEOFENCE = ("NL",) - NO_SUBTITLES = True - - @staticmethod - @click.command(name="PTHS", short_help="https://www.pathe-thuis.nl") - @click.argument("title", type=str) - @click.pass_context - def cli(ctx, **kwargs): - return PTHS(ctx, **kwargs) - - def __init__(self, ctx, title: str): - super().__init__(ctx) - - m = re.match(self.TITLE_RE, title) - if not m: - raise ValueError( - f"Unsupported Pathé Thuis URL or ID: {title}\n" - "Use e.g. https://www.pathe-thuis.nl/film/30591" - ) - - self.movie_id = m.group("id") - self.drm_token = None - - if self.config is None: - raise EnvironmentError("Missing service config for Pathé Thuis.") - - def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None: - super().authenticate(cookies, credential) - - if not cookies: - self.log.warning("No cookies provided, proceeding unauthenticated.") - return - - token = next((c.value for c in cookies if c.name == "authenticationToken"), None) - if not token: - self.log.info("No authenticationToken cookie found, unauthenticated mode.") - return - - self.session.headers.update({ - "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:143.0) Gecko/20100101 Firefox/143.0", - "X-Pathe-Device-Identifier": "web-widevine-1", - "X-Pathe-Auth-Session-Token": token, - }) - self.log.info("Authentication token successfully attached to session.") - - - def get_titles(self) -> Titles_T: - url = self.config["endpoints"]["metadata"].format(movie_id=self.movie_id) - r = self.session.get(url) - r.raise_for_status() - data = r.json() - - movie = Movie( - id_=str(data["id"]), - service=self.__class__, - name=data["name"], - description=data.get("intro", ""), - year=data.get("year"), - language=Language.get(data.get("language", "en")), - data=data, - ) - return Movies([movie]) - - - def get_tracks(self, title: Title_T) -> Tracks: - ticket_id = self._get_ticket_id(title) - url = self.config["endpoints"]["ticket"].format(ticket_id=ticket_id) - - r = self.session.get(url) - r.raise_for_status() - data = r.json() - stream = data["stream"] - - manifest_url = stream.get("url") or stream.get("drmurl") - if not manifest_url: - raise ValueError("No stream manifest URL found.") - - self.drm_token = stream["token"] - self.license_url = stream["rawData"]["licenseserver"] - - tracks = DASH.from_url(manifest_url, session=self.session).to_tracks(language=title.language) - - return tracks - - - def _get_ticket_id(self, title: Title_T) -> str: - """Fetch the user's owned ticket ID if present.""" - data = title.data - for t in (data.get("tickets") or []): - if t.get("playable") and str(t.get("movieId")) == str(self.movie_id): - return str(t["id"]) - raise ValueError("No valid ticket found for this movie. Ensure purchase or login.") - - - def get_chapters(self, title: Title_T): - return [] - - - def get_widevine_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes: - if not self.license_url or not self.drm_token: - raise ValueError("Missing license URL or token.") - - headers = { - "Content-Type": "application/octet-stream", - "Authorization": f"Bearer {self.drm_token}", - } - - params = {"custom_data": self.drm_token} - - r = self.session.post(self.license_url, params=params, data=challenge, headers=headers) - r.raise_for_status() - - if not r.content: - raise ValueError("Empty license response, likely invalid or expired token.") - return r.content \ No newline at end of file