unshackle-services/NPO/__init__.py

import json
import re
from http.cookiejar import CookieJar
from typing import Optional
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 Episode, Movie, Movies, Series, Title_T, Titles_T
from unshackle.core.tracks import Chapter, Tracks, Subtitle


class NPO(Service):
    """
    Service code for NPO Start (npo.nl)
    Version: 1.1.0

    Authorization: optional cookies (free/paid content supported)
    Security: FHD @ L3
              FHD @ SL3000   
              (Widevine and PlayReady support) 

    Supports:
      • Series ↦ https://npo.nl/start/serie/{slug}
      • Movies ↦ https://npo.nl/start/video/{slug}

    Note: Movie inside a series can be downloaded as movie by converting URL to:
          https://npo.nl/start/video/slug

          To change between Widevine and Playready, you need to change the DrmType in config.yaml to either widevine or playready
    """

    TITLE_RE = (
        r"^(?:https?://(?:www\.)?npo\.nl/start/)?"
        r"(?:(?P<type>video|serie)/(?P<slug>[^/]+)"
        r"(?:/afleveringen)?"
        r"(?:/seizoen-(?P<season>[^/]+)/(?P<episode>[^/]+)/afspelen)?)?$"
    )
    GEOFENCE = ("NL",)
    NO_SUBTITLES = False

    @staticmethod
    @click.command(name="NPO", short_help="https://npo.nl")
    @click.argument("title", type=str)
    @click.pass_context
    def cli(ctx, **kwargs):
        return NPO(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 NPO URL: {title}\n"
                "Use /video/slug for movies or /serie/slug for series."
            )

        self.slug = m.group("slug")
        self.kind = m.group("type") or "video"
        self.season_slug = m.group("season")
        self.episode_slug = m.group("episode")

        if self.config is None:
            raise EnvironmentError("Missing service config.")

        # Store CDM reference
        self.cdm = ctx.obj.cdm

    def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
        super().authenticate(cookies, credential)
        if not cookies:
            self.log.info("No cookies, proceeding anonymously.")
            return

        token = next((c.value for c in cookies if c.name == "__Secure-next-auth.session-token"), None)
        if not token:
            self.log.info("No session token, proceeding unauthenticated.")
            return

        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) Firefox/143.0",
            "Origin": "https://npo.nl",
            "Referer": "https://npo.nl/",
        })

        r = self.session.get("https://npo.nl/start/api/domain/user-profiles", cookies=cookies)
        if r.ok and isinstance(r.json(), list) and r.json():
            self.log.info(f"NPO login OK, profiles: {[p['name'] for p in r.json()]}")
        else:
            self.log.warning("NPO auth check failed.")

    def _get_build_id(self, slug: str) -> str:
        """Fetch buildId from the actual video/series page."""
        url = f"https://npo.nl/start/{'video' if self.kind == 'video' else 'serie'}/{slug}"
        r = self.session.get(url)
        r.raise_for_status()
        match = re.search(r'<script id="__NEXT_DATA__" type="application/json">({.*?})</script>', r.text, re.DOTALL)
        if not match:
            raise RuntimeError("Failed to extract __NEXT_DATA__")
        data = json.loads(match.group(1))
        return data["buildId"]

    def get_titles(self) -> Titles_T:
        build_id = self._get_build_id(self.slug)

        if self.kind == "serie":
            url = self.config["endpoints"]["metadata_series"].format(build_id=build_id, slug=self.slug)
        else:
            url = self.config["endpoints"]["metadata"].format(build_id=build_id, slug=self.slug)

        resp = self.session.get(url)
        resp.raise_for_status()
        queries = resp.json()["pageProps"]["dehydratedState"]["queries"]

        def get_data(fragment: str):
            return next((q["state"]["data"] for q in queries if fragment in str(q.get("queryKey", ""))), None)

        if self.kind == "serie":
            series_data = get_data("series:detail-")
            if not series_data:
                raise ValueError("Series metadata not found")

            episodes = []
            seasons = get_data("series:seasons-") or []
            for season in seasons:
                eps = get_data(f"programs:season-{season['guid']}") or []
                for e in eps:
                    episodes.append(
                        Episode(
                            id_=e["guid"],
                            service=self.__class__,
                            title=series_data["title"],
                            season=int(season["seasonKey"]),
                            number=int(e["programKey"]),
                            name=e["title"],
                            description=(e.get("synopsis", {}) or {}).get("long", ""),
                            language=Language.get("nl"),
                            data=e,
                        )
                    )
            return Series(episodes)

        # Movie
        item = get_data("program:detail-") or queries[0]["state"]["data"]
        synopsis = item.get("synopsis", {})
        desc = synopsis.get("long") or synopsis.get("short", "") if isinstance(synopsis, dict) else str(synopsis)
        year = (int(item["firstBroadcastDate"]) // 31536000 + 1970) if item.get("firstBroadcastDate") else None

        return Movies([
            Movie(
                id_=item["guid"],
                service=self.__class__,
                name=item["title"],
                description=desc,
                year=year,
                language=Language.get("nl"),
                data=item,
            )
        ])

    def get_tracks(self, title: Title_T) -> Tracks:
        product_id = title.data.get("productId")
        if not product_id:
            raise ValueError("no productId detected.")

        token_url = self.config["endpoints"]["player_token"].format(product_id=product_id)
        r_tok = self.session.get(token_url, headers={"Referer": f"https://npo.nl/start/video/{self.slug}"})
        r_tok.raise_for_status()
        jwt = r_tok.json()["jwt"]

        # Request stream
        r_stream = self.session.post(
            self.config["endpoints"]["streams"],
            json={
                "profileName": "dash",
                "drmType": self.config["DrmType"],
                "referrerUrl": f"https://npo.nl/start/video/{self.slug}",
                "ster": {"identifier": "npo-app-desktop", "deviceType": 4, "player": "web"},
            },
            headers={
                "Authorization": jwt,
                "Content-Type": "application/json",
                "Origin": "https://npo.nl",
                "Referer": f"https://npo.nl/start/video/{self.slug}",
            },
        )
        r_stream.raise_for_status()
        data = r_stream.json()

        if "error" in data:
            raise PermissionError(f"Stream error: {data['error']}")

        stream = data["stream"]
        manifest_url = stream.get("streamURL") or stream.get("url")
        if not manifest_url:
            raise ValueError("No stream URL in response")

        is_unencrypted = "unencrypted" in manifest_url.lower() or not any(k in stream for k in ["drmToken", "token"])

        # Parse DASH
        tracks = DASH.from_url(manifest_url, session=self.session).to_tracks(language=title.language)

        # Subtitles
        subtitles = []
        for sub in (data.get("assets", {}) or {}).get("subtitles", []) or []:
            if not isinstance(sub, dict):
                continue
            lang = sub.get("iso", "und")
            location = sub.get("location")
            if not location:
                continue  # skip if no URL provided
            subtitles.append(
                Subtitle(
                    id_=sub.get("name", lang),
                    url=location.strip(),
                    language=Language.get(lang),
                    is_original_lang=lang == "nl",
                    codec=Subtitle.Codec.WebVTT,
                    name=sub.get("name", "Unknown"),
                    forced=False,
                    sdh=False,
                )
            )
        tracks.subtitles = subtitles

        # DRM
        if is_unencrypted:
            for tr in tracks.videos + tracks.audio:
                if hasattr(tr, "drm") and tr.drm:
                    tr.drm.clear()
        else:
            self.drm_token = stream.get("drmToken") or stream.get("token") or stream.get("drm_token")
            if not self.drm_token:
                raise ValueError(f"No DRM token found. Available keys: {list(stream.keys())}")

            for tr in tracks.videos + tracks.audio:
                if getattr(tr, "drm", None):
                    if drm_type == "playready":
                        tr.drm.license = lambda challenge, **kw: self.get_playready_license(
                            challenge=challenge, title=title, track=tr
                        )
                    else:
                        tr.drm.license = lambda challenge, **kw: self.get_widevine_license(
                            challenge=challenge, title=title, track=tr
                        )

        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:
        if not self.drm_token:
            raise ValueError("DRM token not set, login or paid content may be required.")
        r = self.session.post(
            self.config["endpoints"]["license"],
            params={"custom_data": self.drm_token},
            data=challenge,
        )
        r.raise_for_status()
        return r.content

    def get_playready_license(self, challenge: bytes, title: Title_T, track: AnyTrack) -> bytes:
        if not self.drm_token:
            raise ValueError("DRM token not set, login or paid content may be required.")
        headers = {
            "Content-Type": "text/xml; charset=utf-8",
            "SOAPAction": "http://schemas.microsoft.com/DRM/2007/03/protocols/AcquireLicense",
            "Origin": "https://npo.nl",
            "Referer": "https://npo.nl/",
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/141.0.0.0 Safari/537.36 Edg/141.0.0.0"
            ),
        }
        r = self.session.post(
            self.config["endpoints"]["license"],
            params={"custom_data": self.drm_token},
            data=challenge,
            headers=headers,
        )
        r.raise_for_status()
        return r.content