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 click import Context
from devine.core.constants import AnyTrack
from devine.core.credential import Credential
from devine.core.manifests import DASH, HLS
from devine.core.search_result import SearchResult
from devine.core.service import Service
from devine.core.titles import Episode, Movie, Movies, Series
from devine.core.tracks import Chapter, Chapters, Tracks
from requests import Request
class CBC(Service):
"""
\b
Service code for CBC Gem streaming service (https://gem.cbc.ca/).
\b
Author: stabbedbybrick
Authorization: Credentials
Robustness:
AES-128: 1080p, DDP5.1
Widevine: 720p, DDP5.1
\b
Tips:
- Input can be complete title URL or just the slug:
SHOW: https://gem.cbc.ca/murdoch-mysteries OR murdoch-mysteries
MOVIE: https://gem.cbc.ca/the-babadook OR the-babadook
\b
Notes:
- The CCExtrator implementation in Devine can't handle this service properly and might fail.
If it does fail, you'll need to remove/comment out the code block in 'devine/commands/dl.py', line 577-616
This will keep CCExtractor from being called at all.
- Some audio tracks contain invalid data, causing warning messages from mkvmerge during muxing
These can be ignored.
"""
GEOFENCE = ("ca",)
ALIASES = ("gem", "cbcgem",)
@staticmethod
@click.command(name="CBC", short_help="https://gem.cbc.ca/", help=__doc__)
@click.argument("title", type=str)
@click.pass_context
def cli(ctx: Context, **kwargs: Any) -> CBC:
return CBC(ctx, **kwargs)
def __init__(self, ctx: Context, title: str):
self.title = title
super().__init__(ctx)
self.base_url = self.config["endpoints"]["base_url"]
def search(self) -> Generator[SearchResult, None, None]:
params = {
"device": "web",
"pageNumber": "1",
"pageSize": "20",
"term": self.title,
}
response = self._request("GET", "/ott/catalog/v1/gem/search", params=params)
for result in response.get("result", []):
yield SearchResult(
id_="https://gem.cbc.ca/{}".format(result.get("url")),
title=result.get("title"),
description=result.get("synopsis"),
label=result.get("type"),
url="https://gem.cbc.ca/{}".format(result.get("url")),
)
def authenticate(self, cookies: Optional[CookieJar] = None, credential: Optional[Credential] = None) -> None:
super().authenticate(cookies, credential)
if not credential:
raise EnvironmentError("Service requires Credentials for Authentication.")
login = self.cache.get(f"login_{credential.sha1}")
tokens = self.cache.get(f"tokens_{credential.sha1}")
if login and not login.expired:
# cached
self.log.info(" + Using cached login tokens")
auth_token = login.data["access_token"]
elif login and login.expired:
payload = {
"email": credential.username,
"password": credential.password,
"refresh_token": login.data["refresh_token"],
}
params = {"apikey": self.config["endpoints"]["api_key"]}
auth = self._request(
"POST", "https://api.loginradius.com/identity/v2/auth/login",
payload=payload,
params=params,
)
login.set(auth, expiration=auth["expires_in"])
auth_token = login.data["access_token"]
self.log.info(" + Refreshed login tokens")
else:
payload = {
"email": credential.username,
"password": credential.password,
}
params = {"apikey": self.config["endpoints"]["api_key"]}
auth = self._request(
"POST", "https://api.loginradius.com/identity/v2/auth/login",
payload=payload,
params=params,
)
login.set(auth, expiration=auth["expires_in"])
auth_token = login.data["access_token"]
self.log.info(" + Acquired fresh login tokens")
if tokens and not tokens.expired:
# cached
self.log.info(" + Using cached access tokens")
access_token = tokens.data["accessToken"]
else:
access = self.access_token(auth_token)
tokens.set(access, expiration=access["accessTokenExpiresIn"])
access_token = access["accessToken"]
self.log.info(" + Acquired fresh access tokens")
claims_token = self.claims_token(access_token)
self.session.headers.update({"x-claims-token": claims_token})
def get_titles(self) -> Union[Movies, Series]:
title_re = r"^(?:https?://(?:www.)?gem.cbc.ca/)?(?P<id>[a-zA-Z0-9_-]+)"
try:
title_id = re.match(title_re, self.title).group("id")
except Exception:
raise ValueError("- Could not parse ID from title")
data = self._request("GET", "/ott/cbc-api/v2/shows/{}".format(title_id))
label = data.get("seasons", [])[0].get("title")
if label.lower() in ("film", "movie"):
movie = self._movie(data)
return Movies(movie)
else:
episodes = self._show(data)
return Series(episodes)
def get_tracks(self, title: Union[Movie, Episode]) -> Tracks:
media_id = title.data["playSession"].get("mediaId")
index = self._request(
"GET", "/media/meta/v1/index.ashx",
params={"appCode": "gem", "idMedia": media_id, "output": "jsonObject"}
)
title.data["extra"] = {
"chapters": index["Metas"].get("Chapitres"),
"credits": index["Metas"].get("CreditStartTime"),
}
self.drm = index["Metas"].get("isDrmActive") == "true"
if self.drm:
tech = next(tech["name"] for tech in index["availableTechs"] if "widevine" in tech["drm"])
else:
tech = next(tech["name"] for tech in index["availableTechs"] if not tech["drm"])
response = self._request(
"GET", self.config["endpoints"]["validation"].format("android", media_id, "smart-tv", tech)
)
manifest = response.get("url")
self.license = next((x["value"] for x in response["params"] if "widevineLicenseUrl" in x["name"]), None)
self.token = next((x["value"] for x in response["params"] if "widevineAuthToken" in x["name"]), None)
stream_type = HLS if tech == "hls" else DASH
tracks = stream_type.from_url(manifest, self.session).to_tracks(language=index.get("Language", "en"))
if stream_type == DASH:
for track in tracks.audio:
label = track.data["dash"]["adaptation_set"].find("Label")
if label is not None and "descriptive" in label.text.lower():
track.descriptive = True
return tracks
def get_chapters(self, title: Union[Movie, Episode]) -> Chapters:
extra = title.data["extra"]
chapters = []
if extra.get("chapters"):
chapters = [Chapter(timestamp=x) for x in extra["chapters"].split(",")]
if extra.get("credits"):
chapters.append(Chapter(name="Credits", timestamp=float(extra["credits"])))
return Chapters(chapters)
def get_widevine_service_certificate(self, **_: Any) -> str:
return None
def get_widevine_license(
self, *, challenge: bytes, title: Union[Movies, Series], track: AnyTrack
) -> Optional[Union[bytes, str]]:
if not self.license or not self.token:
return None
headers = {"x-dt-auth-token": self.token}
r = self.session.post(self.license, headers=headers, data=challenge)
r.raise_for_status()
return r.content
# Service specific
def _show(self, data: dict) -> Episode:
episodes = [episode for season in data["seasons"] for episode in season["assets"] if not episode["isTrailer"]]
return Series(
[
Episode(
id_=episode["id"],
service=self.__class__,
title=data.get("title"),
season=int(episode.get("season", 0)),
number=int(episode.get("episode", 0)),
name=episode.get("title"),
data=episode,
)
for episode in episodes
]
)
def _movie(self, data: dict) -> Movie:
movies = [movie for season in data["seasons"] for movie in season["assets"] if not movie["isTrailer"]]
return [
Movie(
id_=movie.get("id"),
service=self.__class__,
name=data.get("title"),
data=movie,
)
for movie in movies
]
def access_token(self, token: str) -> str:
params = {
"access_token": token,
"apikey": self.config["endpoints"]["api_key"],
"jwtapp": "jwt",
}
headers = {"content-type": "application/json"}
resp = self._request(
"GET", "https://cloud-api.loginradius.com/sso/jwt/api/token",
headers=headers,
params=params
)
payload = {"jwt": resp.get("signature")}
headers = {"content-type": "application/json", "ott-device-type": "web"}
auth = self._request("POST", "/ott/cbc-api/v2/token", headers=headers, payload=payload)
return auth
def claims_token(self, token: str) -> str:
headers = {
"content-type": "application/json",
"ott-device-type": "web",
"ott-access-token": token,
}
response = self._request("GET", "/ott/cbc-api/v2/profile", headers=headers)
return response["claimsToken"]
def _request(
self, method: str, api: str, params: dict = None, headers: dict = None, payload: dict = None
) -> Any[dict | str]:
url = urljoin(self.base_url, api)
headers = headers or self.session.headers
prep = self.session.prepare_request(Request(method, url, params=params, headers=headers, json=payload))
response = self.session.send(prep)
if response.status_code not in (200, 426):
raise ConnectionError(f"{response.status_code} - {response.text}")
try:
data = json.loads(response.content)
error_keys = ["errorMessage", "ErrorMessage", "ErrorCode", "errorCode", "error"]
error_message = next((data.get(key) for key in error_keys if key in data), None)
if error_message:
self.log.error(f"\n - Error: {error_message}\n")
sys.exit(1)
return data
except json.JSONDecodeError:
raise ConnectionError("Request for {} failed: {}".format(response.url, response.text))