293 lines
11 KiB
Python
293 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import sys
|
|
from collections.abc import Generator
|
|
from http.cookiejar import CookieJar
|
|
from typing import Any, Optional, Union
|
|
from urllib.parse import urljoin
|
|
|
|
import click
|
|
from requests import Request
|
|
|
|
from devine.core.constants import AnyTrack
|
|
from devine.core.credential import Credential
|
|
from devine.core.manifests import DASH
|
|
from devine.core.search_result import SearchResult
|
|
from devine.core.service import Service
|
|
from devine.core.titles import Episode, Movie, Movies, Series, Title_T, Titles_T
|
|
from devine.core.tracks import Chapter, Chapters, Tracks
|
|
from devine.core.utils.sslciphers import SSLCiphers
|
|
from devine.core.utils.xml import load_xml
|
|
|
|
|
|
class PMTP(Service):
|
|
"""
|
|
Service code for Paramount's Paramount+ streaming service (https://paramountplus.com).
|
|
|
|
\b
|
|
Author: stabbedbybrick
|
|
Authorization: None for US | Credentials for INTL
|
|
Robustness:
|
|
Widevine:
|
|
L3: 2160p, DDP5.1
|
|
|
|
\b
|
|
Tips:
|
|
- Input should be complete URLs for both shows and movies:
|
|
https://www.paramountplus.com/movies/video/3vxCeaSHnqJLmatpwS1OzrkdA16h7sN9/
|
|
https://www.paramountplus.com/shows/special-ops-lioness/
|
|
- Use -r, --region to specify your region. If not used, the default is "us".
|
|
|
|
\b
|
|
Notes:
|
|
- Credentials are only required for INTL.
|
|
"""
|
|
|
|
ALIASES = ("paramountplus", "paramount+")
|
|
# GEOFENCE = ("us",)
|
|
TITLE_RE = r"https://www\.paramountplus\.com(/.*?)?/(?P<type>shows|movies)(/(?P<video>.*?))?/(?P<id>[a-zA-Z0-9_-]+)"
|
|
|
|
@staticmethod
|
|
@click.command(name="PMTP", short_help="https://paramountplus.com")
|
|
@click.argument("title", type=str, required=False)
|
|
@click.option("-r", "--region", default="us", help="Specify region (default: US)")
|
|
@click.pass_context
|
|
def cli(ctx, **kwargs):
|
|
return PMTP(ctx, **kwargs)
|
|
|
|
def __init__(self, ctx, title: str, region: str):
|
|
self.title = title
|
|
self.region = region.lower()
|
|
self.GEOFENCE = (self.region,)
|
|
super().__init__(ctx)
|
|
|
|
self.endpoints = self.config["endpoints"].get(self.region, self.config["endpoints"]["intl"])
|
|
|
|
def search(self) -> Generator[SearchResult, None, None]:
|
|
params = {
|
|
"term": self.title,
|
|
"termCount": 50,
|
|
"showCanVids": "true",
|
|
}
|
|
results = self._request("GET", "/apps-api/v3.1/androidphone/contentsearch/search.json", params=params)["terms"]
|
|
|
|
for result in results:
|
|
yield SearchResult(
|
|
id_=result.get("path"),
|
|
title=result.get("title"),
|
|
description=None,
|
|
label=result.get("term_type"),
|
|
url=result.get("path"),
|
|
)
|
|
|
|
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
|
|
super().authenticate(cookies, credential)
|
|
if self.region != "us":
|
|
if not credential:
|
|
raise EnvironmentError("INTL requires Credentials for Authentication.")
|
|
|
|
params = {
|
|
"j_username": credential.username,
|
|
"j_password": credential.password,
|
|
}
|
|
self._request("POST", "/apps-api/v2.0/androidphone/auth/login.json", params=params)
|
|
|
|
status = self._request("GET", "/apps-api/v3.0/androidphone/login/status.json")
|
|
if not status.get("isLoggedIn"):
|
|
raise ConnectionError(" - Failed to authenticate user. Credentials may be invalid.")
|
|
|
|
def get_titles(self) -> Titles_T:
|
|
try:
|
|
kind, video, title_id = (re.match(self.TITLE_RE, self.title).group(i) for i in ("type", "video", "id"))
|
|
except Exception:
|
|
self.log.error("- Could not parse ID from title")
|
|
sys.exit(1)
|
|
|
|
if kind == "movies":
|
|
movies = self._movie(title_id)
|
|
return Movies(movies)
|
|
|
|
elif video:
|
|
episodes = self._episode(title_id)
|
|
return Series(episodes)
|
|
|
|
elif kind == "shows":
|
|
episodes = self._show(title_id)
|
|
return Series(episodes)
|
|
|
|
def get_tracks(self, title: Title_T) -> Tracks:
|
|
self.token, self.license = self.ls_session(title.id)
|
|
manifest = self.get_manifest(title)
|
|
tracks = DASH.from_url(manifest, self.session).to_tracks(language=title.language)
|
|
return tracks
|
|
|
|
def get_chapters(self, title: Episode) -> Chapters:
|
|
if not title.data.get("playbackEvents", {}).get("endCreditChapterTimeMs"):
|
|
return Chapters()
|
|
|
|
end_credits = title.data["playbackEvents"]["endCreditChapterTimeMs"]
|
|
return Chapters([Chapter(name="Credits", timestamp=end_credits)])
|
|
|
|
def certificate(self, *, challenge: bytes, title: Title_T, track: AnyTrack):
|
|
return self.session.post(self.license, data=challenge).content
|
|
|
|
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
|
|
headers = {"Authorization": f"Bearer {self.token}"}
|
|
r = self.session.post(self.license, headers=headers, data=challenge)
|
|
if not r.ok:
|
|
self.log.error(r.text)
|
|
sys.exit(1)
|
|
return r.content
|
|
|
|
# Service specific functions
|
|
|
|
|
|
def _movie(self, title: str) -> Movie:
|
|
res = self._request(
|
|
"GET", "/apps-api/v3.0/androidphone/movies/{}.json".format(title),
|
|
params={"includeContentInfo": "true"}
|
|
)
|
|
title = res["movie"]["movieContent"]
|
|
return [
|
|
Movie(
|
|
id_=title["contentId"],
|
|
name=title["title"],
|
|
year=title["_airDateISO"][:4],
|
|
language=res["movie"].get("locale", "en-US"),
|
|
service=self.__class__,
|
|
data=title,
|
|
)
|
|
]
|
|
|
|
def _show(self, title: str) -> Episode:
|
|
data = self._request("GET", "/apps-api/v3.0/androidphone/shows/slug/{}.json".format(title))
|
|
|
|
links = next((x.get("links") for x in data["showMenu"] if x.get("device_app_id") == "all_platforms"), None)
|
|
config = next((x.get("videoConfigUniqueName") for x in links if x.get("title").strip() == "Episodes"), None)
|
|
show = next((x for x in data["show"]["results"] if x.get("type") == "show"), None)
|
|
seasons = [x["seasonNum"] for x in data["available_video_seasons"]["itemList"] if x.get("seasonNum")]
|
|
locale = show.get("locale", "en-US")
|
|
|
|
show_data = self._request(
|
|
"GET",
|
|
"/apps-api/v2.0/androidphone/shows/{}/videos/config/{}.json".format(show.get("show_id"), config),
|
|
params={"platformType": "apps", "rows": "1", "begin": "0"},
|
|
)
|
|
|
|
section = next(
|
|
(x["sectionId"] for x in show_data["videoSectionMetadata"] if x["title"] == "Full Episodes"), None
|
|
)
|
|
|
|
episodes = []
|
|
for season in seasons:
|
|
res = self._request(
|
|
"GET",
|
|
"/apps-api/v2.0/androidphone/videos/section/{}.json".format(section),
|
|
params={"begin": "0", "rows": "999", "params": f"seasonNum={season}", "seasonNum": season},
|
|
)
|
|
episodes.extend(res["sectionItems"].get("itemList"))
|
|
|
|
return [
|
|
Episode(
|
|
id_=episode["contentId"],
|
|
title=episode["seriesTitle"],
|
|
season=episode["seasonNum"] if episode["fullEpisode"] else 0,
|
|
number=episode["episodeNum"] if episode["fullEpisode"] else episode["positionNum"],
|
|
name=episode["label"],
|
|
language=locale,
|
|
service=self.__class__,
|
|
data=episode,
|
|
)
|
|
for episode in episodes
|
|
if episode["fullEpisode"]
|
|
]
|
|
|
|
def _episode(self, title: str) -> Episode:
|
|
data = self._request("GET", "/apps-api/v2.0/androidphone/video/cid/{}.json".format(title))
|
|
|
|
return [
|
|
Episode(
|
|
id_=episode["contentId"],
|
|
title=episode["seriesTitle"],
|
|
season=episode["seasonNum"] if episode["fullEpisode"] else 0,
|
|
number=episode["episodeNum"] if episode["fullEpisode"] else episode["positionNum"],
|
|
name=episode["label"],
|
|
language=episode.get("locale", "en-US"),
|
|
service=self.__class__,
|
|
data=episode,
|
|
)
|
|
for episode in data["itemList"]
|
|
]
|
|
|
|
def ls_session(self, content_id: str) -> str:
|
|
res = self._request(
|
|
"GET",
|
|
"/apps-api/v3.1/androidphone/irdeto-control/anonymous-session-token.json",
|
|
params={"contentId": content_id},
|
|
)
|
|
|
|
return res.get("ls_session"), res.get("url")
|
|
|
|
def get_manifest(self, title: Episode) -> str:
|
|
try:
|
|
res = self._request(
|
|
"GET",
|
|
"http://link.theplatform.com/s/{}/media/guid/2198311517/{}".format(
|
|
title.data.get("cmsAccountId"), title.id
|
|
),
|
|
params={
|
|
"format": "SMIL",
|
|
"assetTypes": "|".join(self.config["assets"]),
|
|
"formats": "MPEG-DASH,MPEG4,M3U",
|
|
},
|
|
)
|
|
|
|
body = load_xml(res).find("body").find("seq").findall("switch")
|
|
bitrate = max(body, key=lambda x: int(x.find("video").get("system-bitrate")))
|
|
videos = [x.get("src") for x in bitrate.findall("video")]
|
|
if not videos:
|
|
raise ValueError("Could not find any streams - is the title still available?")
|
|
|
|
manifest = next(
|
|
(x for x in videos if "hdr_dash" in x.lower()),
|
|
next((x for x in videos if "cenc_dash" in x.lower()), videos[0]),
|
|
)
|
|
|
|
except Exception as e:
|
|
self.log.warning("ThePlatform request failed: {}, falling back to streaming manifest".format(e))
|
|
if not title.data.get("streamingUrl"):
|
|
raise ValueError("Could not find any streams - is the title still available?")
|
|
|
|
manifest = title.data.get("streamingUrl")
|
|
|
|
return manifest
|
|
|
|
def _request(self, method: str, api: str, params: dict = None, headers: dict = None) -> Any[dict | str]:
|
|
url = urljoin(self.endpoints["base_url"], api)
|
|
self.session.headers.update(self.config["headers"])
|
|
self.session.params = {"at": self.endpoints["token"]}
|
|
#for prefix in ("https://", "http://"):
|
|
# self.session.mount(prefix, SSLCiphers(security_level=2))
|
|
|
|
if params:
|
|
self.session.params.update(params)
|
|
if headers:
|
|
self.session.headers.update(headers)
|
|
|
|
prep = self.session.prepare_request(Request(method, url))
|
|
|
|
response = self.session.send(prep)
|
|
if response.status_code != 200:
|
|
raise ConnectionError(f"{response.text}")
|
|
|
|
try:
|
|
data = json.loads(response.content)
|
|
if not data.get("success"):
|
|
raise ValueError(data.get("message"))
|
|
return data
|
|
|
|
except json.JSONDecodeError:
|
|
return response.text
|