diff --git a/Install-media-tools.ps1 b/Install-media-tools.ps1 new file mode 100644 index 0000000..1e51f4f --- /dev/null +++ b/Install-media-tools.ps1 @@ -0,0 +1,1452 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + envied/Install-media-tools.ps1 at main · vinefeeder/envied · GitHub + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+ + + + +
+ Skip to content + + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + + + + +
+ +
+ + + + + + + + +
+ + + + + +
+ + + + + + + + + +
+
+
+ + + + + + + + + + +
+ +
+ +
+ +
+ + + + / + + envied + + + Public +
+ + +
+ +
+ + +
+
+ +
+
+ + + + +
+ + + + + + + +
+ + + + + + + + + + + + + + + + + + +
+ + + + + +
+ + + +
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + +
+
+
+ + + diff --git a/Install-media-tools.sh b/Install-media-tools.sh new file mode 100644 index 0000000..99225b0 --- /dev/null +++ b/Install-media-tools.sh @@ -0,0 +1,1452 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + envied/Install-media-tools.sh at main · vinefeeder/envied · GitHub + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+
+ + + + +
+ Skip to content + + + + + + + + + + + +
+ + + + + + + + + + + + + + + + + + + + +
+ +
+ + + + + + + + +
+ + + + + +
+ + + + + + + + + +
+
+
+ + + + + + + + + + +
+ +
+ +
+ +
+ + + + / + + envied + + + Public +
+ + +
+ +
+ + +
+
+ +
+
+ + + + +
+ + + + + + + +
+ + + + + + + + + + + + + + + + + + +
+ + + + + +
+ + + +
+
+ +
+ + + + + + + + + + + + + + + + + + + + + + +
+
+
+ + + diff --git a/README.md b/README.md index 686affa..41be60e 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ uv run vinefeeder - to open the interactive GUI for search, browse, etc. To go to the command line use of 'envied' after using the GUI: close the GUI directly, or ctrl+c in terminal. ``` -**Installation** +**Installation** - with binaries already installed Uv is the package manager and loads both VineFeeder and Envied together. Envied runs independenly or may be called by Vinefeeder. @@ -50,6 +50,67 @@ uv sync uv run vinefeeder --help or uv run envied --help ``` +**Installation** with a bare machine. + +Find and install git: +* Windows: + download https://github.com/git-for-windows/git/releases/download/v2.52.0.windows.1/Git-2.52.0-64-bit.exe and run the installer +* Linux: + install from your OS' package manager. +Find and install Python: +* Windows: + search Windows Store for Python and select the latest version available 3.12 or above. +* Linux: + Python is probably already installed in basic form. + +Find and install TwinVine +* Windows or Linux: + inside a command window (cmd -w typed in start for windows or a terminal for Linux) + ``` + git clone https://github.com/vinefeeder/TwinVine.git + ``` + A folder containing all the TwinVine files will be created. Close the command window but note the location of TwinVine. + +Install all the required binary files and add to system variable - Path. + +* Windows + - Open Start + - Type PowerShell + - Right-click Windows PowerShell → Run as administrator + - Inside PowerShell change directory to TwinVine and run the following command + + + ``` + powershell -ExecutionPolicy Bypass -File .\Install-media-tools.ps1 + + ``` +* Linux: + - Open a terminal window, cd to TwinVine and run + ``` + sudo bash Install-media-tools.sh + ``` +* Windows: +Once installation of all the binaries has finished close the PowerShell window with admininstrator privileges and open an ordinary PowerShell window. Change directory to Twinvine + + ``` + uv clean + uv lock + uv sync + uv run vinefeeder --help or + uv run envied --help + ``` +* Linux: + - Once the binary install has finished navigate to TwinVine top level folder and run the following commands one by one to build the TwinVine system. + + ``` + uv clean + uv lock + uv sync + uv run vinefeeder --help or + uv run envied --help + ``` + + **Configuration** Run this line inside the TwinVine folder: diff --git a/packages/envied/src/envied/services/DSCP/__init__.py b/packages/envied/src/envied/services/DSCP/__init__.py index f8643b9..de98633 100644 --- a/packages/envied/src/envied/services/DSCP/__init__.py +++ b/packages/envied/src/envied/services/DSCP/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import re +import sys import uuid from collections import defaultdict from collections.abc import Generator @@ -32,9 +33,9 @@ class DSCP(Service): Credit to @sp4rk.y for the subtitle fix. \b - Version: 1.0.0 + Version: 1.0.1 Author: stabbedbybrick - Authorization: Cookies + Authorization: Cookies for subscription, none for freely available titles Robustness: Widevine: L1: 2160p, 1080p @@ -45,11 +46,11 @@ class DSCP(Service): \b Tips: - - Input examples: - SHOW: https://play.discoveryplus.com/show/eb26e00e-9582-4790-a61c-48d785926f58 - STANDALONE: https://play.discoveryplus.com/standalone/5012ae3f-d9bd-46ec-ad42-b8116b811441 - SPORT: https://play.discoveryplus.com/sport/9cc449de-2a64-524d-bcb6-cabd4ac70340 - EPISODE: https://play.discoveryplus.com/video/watch/8685efdd-a3c4-4892-b1d1-5f9f071cacf1/de67ea8e-a90f-4609-81af-4f09906f60b2 + - Input can be either complete title URL or just the path: + SHOW: /show/eb26e00e-9582-4790-a61c-48d785926f58 + STANDALONE: /standalone/5012ae3f-d9bd-46ec-ad42-b8116b811441 + SPORT: /sport/9cc449de-2a64-524d-bcb6-cabd4ac70340 + EPISODE: /video/watch/8685efdd-a3c4-4892-b1d1-5f9f071cacf1/de67ea8e-a90f-4609-81af-4f09906f60b2 \b Notes: @@ -93,30 +94,32 @@ class DSCP(Service): self.drm_system = "widevine" self.security_level = "L1" + self.base_url = self.config["endpoints"]["default_url"] + def get_session(self) -> CurlSession: return CurlSession("okhttp4", status_forcelist=[429, 502, 503, 504]) def authenticate(self, cookies: CookieJar | None = None, credential: Credential | None = None) -> None: super().authenticate(cookies, credential) - if not cookies: - raise EnvironmentError("Service requires Cookies for Authentication.") + tokens = {} - cache = self.cache.get(f"tokens_{self.profile}") - if cache: - self.log.info(" + Using cached Tokens...") - tokens = cache.data - else: - self.log.info(" + Setting up new profile...") - - st_token = self.session.cookies.get_dict().get("st") + if cookies is not None: + st_token = next((c.value for c in cookies if c.name == "st"), None) if not st_token: raise ValueError("- Unable to find token in cookies, try refreshing.") - profile = {"token": st_token, "device_id": str(uuid.uuid1())} - cache.set(profile) - tokens = cache.data + # Only use cache if cookies are present since it's not needed for free titles + cache = self.cache.get(f"tokens_{self.profile}") + if cache: + self.log.info(" + Using cached Tokens...") + tokens = cache.data + else: + self.log.info(" + Setting up new profile...") + profile = {"token": st_token, "device_id": str(uuid.uuid1())} + cache.set(profile) + tokens = cache.data - self.device_id = tokens["device_id"] + self.device_id = tokens.get("device_id") or str(uuid.uuid1()) client_id = self.config["client_id"] self.session.headers.update({ @@ -126,17 +129,12 @@ class DSCP(Service): "x-device-info": f"dplus/20.8.1.2 (NVIDIA/SHIELD Android TV; android/9-mdarcy; {self.device_id}/{client_id})", }) - self.base_url = self.config["endpoints"]["base_url"].format("any", "any") access = self._request("GET", "/token", params={"realm": "bolt", "deviceId": self.device_id}) self.access_token = access["data"]["attributes"]["token"] config = self._request("POST", "/session-context/headwaiter/v1/bootstrap") - self.base_url = self.config["endpoints"]["base_url"].format(config["routing"]["tenant"], config["routing"]["homeMarket"]) - - user = self._request("GET", "/users/me") - if user["data"]["attributes"]["anonymous"]: - raise ValueError("Anonymous user - try refreshing cookies.") + self.base_url = self.config["endpoints"]["template"].format(config["routing"]["tenant"], config["routing"]["homeMarket"]) def search(self) -> Generator[SearchResult, None, None]: params = { @@ -152,11 +150,11 @@ class DSCP(Service): for result in results: yield SearchResult( - id_=f"https://play.discoveryplus.com/show/{result.get('alternateId')}", + id_=f"/show/{result.get('alternateId')}", title=result.get("name"), description=result.get("description"), label="show", - url=f"https://play.discoveryplus.com/show/{result.get('alternateId')}", + url=f"/show/{result.get('alternateId')}", ) def get_titles(self) -> Movies | Series: @@ -309,7 +307,7 @@ class DSCP(Service): elif "credits" in chapter.get("type", "").lower(): chapters.append(Chapter(name="Credits", timestamp=chapter["start"])) - if not any(c.timestamp == 0 for c in chapters): + if not any(c.timestamp == "00:00:00.000" for c in chapters): chapters.append(Chapter(timestamp=0)) return sorted(chapters, key=lambda x: x.timestamp) @@ -574,8 +572,11 @@ class DSCP(Service): try: data = json.loads(response.content) - if data.get("errors"): - raise ConnectionError(data["errors"]) + if errors := data.get("errors", []): + code = next((x.get("code", "") for x in errors), "") + if "missingpackage" in code.lower(): + self.log.error("\nError: Subscription is required for this title.") + sys.exit(1) return data diff --git a/packages/envied/src/envied/services/DSCP/config.yaml b/packages/envied/src/envied/services/DSCP/config.yaml index 700cd9e..9092b37 100644 --- a/packages/envied/src/envied/services/DSCP/config.yaml +++ b/packages/envied/src/envied/services/DSCP/config.yaml @@ -1,4 +1,5 @@ endpoints: - base_url: "https://default.{}-{}.prd.api.discoveryplus.com" + template: "https://default.{}-{}.prd.api.discoveryplus.com" + default_url: "https://default.any-any.prd.api.discoveryplus.com" client_id: "b6746ddc-7bc7-471f-a16c-f6aaf0c34d26" # androidtv diff --git a/packages/envied/src/envied/services/RTE/__init__.py b/packages/envied/src/envied/services/RTE/__init__.py index 1ca5636..4098bb1 100644 --- a/packages/envied/src/envied/services/RTE/__init__.py +++ b/packages/envied/src/envied/services/RTE/__init__.py @@ -24,7 +24,7 @@ class RTE(Service): Service code for RTE Player streaming service (https://www.rte.ie/player/). \b - Version: 1.0.1 + Version: 1.0.3 Author: stabbedbybrick Authorization: None Robustness: @@ -97,7 +97,7 @@ class RTE(Service): title_re = ( r"https://www\.rte\.ie/player" r"/(?Pseries|movie)" - r"/(?P[a-zA-Z0-9_-]+)" + r"/(?P[a-zA-Z0-9%_.-]+)" r"/(?P[a-zA-Z0-9_\-=?]+)/?$" ) try: diff --git a/packages/envied/src/envied/services/services/ALL4/__init__.py b/packages/envied/src/envied/services/services/ALL4/__init__.py new file mode 100644 index 0000000..da3e2c8 --- /dev/null +++ b/packages/envied/src/envied/services/services/ALL4/__init__.py @@ -0,0 +1,419 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +import re +import sys +from collections.abc import Generator +from datetime import datetime, timezone +from http.cookiejar import MozillaCookieJar +from typing import Any, Optional, Union + +import click +from click import Context +from Crypto.Util.Padding import unpad +from Cryptodome.Cipher import AES +from pywidevine.cdm import Cdm as WidevineCdm +from envied.core.credential import Credential +from envied.core.manifests.dash import DASH +from envied.core.search_result import SearchResult +from envied.core.service import Service +from envied.core.titles import Episode, Movie, Movies, Series +from envied.core.tracks import Chapter, Subtitle, Tracks + + +class ALL4(Service): + """ + Service code for Channel 4's All4 streaming service (https://channel4.com). + + \b + Version: 1.0.1 + Author: stabbedbybrick + Authorization: Credentials + Robustness: + L3: 1080p, AAC2.0 + + \b + Tips: + - Use complete title URL or slug as input: + https://www.channel4.com/programmes/taskmaster OR taskmaster + - Use on demand URL for directly downloading episodes: + https://www.channel4.com/programmes/taskmaster/on-demand/75588-002 + - Both android and web/pc endpoints are checked for quality profiles. + If android is missing 1080p, it automatically falls back to web. + """ + + GEOFENCE = ("gb", "ie") + TITLE_RE = r"^(?:https?://(?:www\.)?channel4\.com/programmes/)?(?P[a-z0-9-]+)(?:/on-demand/(?P[0-9-]+))?" + + @staticmethod + @click.command(name="ALL4", short_help="https://channel4.com", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> ALL4: + return ALL4(ctx, **kwargs) + + def __init__(self, ctx: Context, title: str): + self.title = title + super().__init__(ctx) + + self.authorization: str + self.asset_id: int + self.license_token: str + self.manifest: str + + self.session.headers.update( + { + "X-C4-Platform-Name": self.config["device"]["platform_name"], + "X-C4-Device-Type": self.config["device"]["device_type"], + "X-C4-Device-Name": self.config["device"]["device_name"], + "X-C4-App-Version": self.config["device"]["app_version"], + "X-C4-Optimizely-Datafile": self.config["device"]["optimizely_datafile"], + } + ) + + def authenticate(self, cookies: Optional[MozillaCookieJar] = None, credential: Optional[Credential] = None) -> None: + super().authenticate(cookies, credential) + if not credential: + raise EnvironmentError("Service requires Credentials for Authentication.") + + cache = self.cache.get(f"tokens_{credential.sha1}") + + if cache and not cache.expired: + # cached + self.log.info(" + Using cached Tokens...") + tokens = cache.data + elif cache and cache.expired: + # expired, refresh + self.log.info("Refreshing cached Tokens") + r = self.session.post( + self.config["endpoints"]["login"], + headers={"authorization": f"Basic {self.config['android']['auth']}"}, + data={ + "grant_type": "refresh_token", + "username": credential.username, + "password": credential.password, + "refresh_token": cache.data["refreshToken"], + }, + ) + try: + res = r.json() + except json.JSONDecodeError: + raise ValueError(f"Failed to refresh tokens: {r.text}") + + if "error" in res: + self.log.error(f"Failed to refresh tokens: {res['errorMessage']}") + sys.exit(1) + + tokens = res + self.log.info(" + Refreshed") + else: + # new + headers = {"authorization": f"Basic {self.config['android']['auth']}"} + data = { + "grant_type": "password", + "username": credential.username, + "password": credential.password, + } + r = self.session.post(self.config["endpoints"]["login"], headers=headers, data=data) + try: + res = r.json() + except json.JSONDecodeError: + raise ValueError(f"Failed to log in: {r.text}") + + if "error" in res: + self.log.error(f"Failed to log in: {res['errorMessage']}") + sys.exit(1) + + tokens = res + self.log.info(" + Acquired tokens...") + + cache.set(tokens, expiration=tokens["expiresIn"]) + + self.authorization = f"Bearer {tokens['accessToken']}" + + def search(self) -> Generator[SearchResult, None, None]: + params = { + "expand": "default", + "q": self.title, + "limit": "100", + "offset": "0", + } + + r = self.session.get(self.config["endpoints"]["search"], params=params) + r.raise_for_status() + + results = r.json() + if isinstance(results["results"], list): + for result in results["results"]: + yield SearchResult( + id_=result["brand"].get("websafeTitle"), + title=result["brand"].get("title"), + description=result["brand"].get("description"), + label=result.get("label"), + url=result["brand"].get("href"), + ) + + def get_titles(self) -> Union[Movies, Series]: + title, on_demand = (re.match(self.TITLE_RE, self.title).group(i) for i in ("id", "vid")) + + r = self.session.get( + self.config["endpoints"]["title"].format(title=title), + params={"client": "android-mod", "deviceGroup": "mobile", "include": "extended-restart"}, + headers={"Authorization": self.authorization}, + ) + if not r.ok: + self.log.error(r.text) + sys.exit(1) + + data = r.json() + + if on_demand is not None: + episodes = [ + Episode( + id_=episode["programmeId"], + service=self.__class__, + title=data["brand"]["title"], + season=episode["seriesNumber"], + number=episode["episodeNumber"], + name=episode["originalTitle"], + language="en", + data=episode["assetInfo"].get("streaming") or episode["assetInfo"].get("download"), + ) + for episode in data["brand"]["episodes"] + if episode.get("assetInfo") and episode["programmeId"] == on_demand + ] + if not episodes: + # Parse HTML of episode page to find title + data = self.get_html(self.title) + episodes = [ + Episode( + id_=data["selectedEpisode"]["programmeId"], + service=self.__class__, + title=data["brand"]["title"], + season=data["selectedEpisode"]["seriesNumber"] or 0, + number=data["selectedEpisode"]["episodeNumber"] or 0, + name=data["selectedEpisode"]["originalTitle"], + language="en", + data=data["selectedEpisode"], + ) + ] + + return Series(episodes) + + elif data["brand"]["programmeType"] == "FM": + return Movies( + [ + Movie( + id_=movie["programmeId"], + service=self.__class__, + name=data["brand"]["title"], + year=int(data["brand"]["summary"].split(" ")[0].strip().strip("()")), + language="en", + data=movie["assetInfo"].get("streaming") or movie["assetInfo"].get("download"), + ) + for movie in data["brand"]["episodes"] + ] + ) + else: + return Series( + [ + Episode( + id_=episode["programmeId"], + service=self.__class__, + title=data["brand"]["title"], + season=episode["seriesNumber"], + number=episode["episodeNumber"], + name=episode["originalTitle"], + language="en", + data=episode["assetInfo"].get("streaming") or episode["assetInfo"].get("download"), + ) + for episode in data["brand"]["episodes"] + if episode.get("assetInfo") + ] + ) + + def get_tracks(self, title: Union[Movie, Episode]) -> Tracks: + android_assets: tuple = self.android_playlist(title.id) + web_assets: tuple = self.web_playlist(title.id) + self.manifest, self.license_token, subtitle, data = self.sort_assets(title, android_assets, web_assets) + self.asset_id = int(title.data["assetId"]) + + tracks = DASH.from_url(self.manifest, self.session).to_tracks(title.language) + tracks.videos[0].data = data + + # manifest subtitles are sometimes empty even if they exist + # so we clear them and add the subtitles manually + tracks.subtitles.clear() + if subtitle is not None: + tracks.add( + Subtitle( + id_=hashlib.md5(subtitle.encode()).hexdigest()[0:6], + url=subtitle, + codec=Subtitle.Codec.from_mime(subtitle[-3:]), + language=title.language, + is_original_lang=True, + forced=False, + sdh=False, + ) + ) + else: + self.log.warning("- Subtitles are either missing or empty") + + for track in tracks.audio: + role = track.data["dash"]["representation"].find("Role") + if role is not None and role.get("value") in ["description", "alternative", "alternate"]: + track.descriptive = True + + return tracks + + def get_chapters(self, title: Union[Movie, Episode]) -> list[Chapter]: + track = title.tracks.videos[0] + + chapters = [ + Chapter( + name=f"Chapter {i + 1:02}", + timestamp=datetime.fromtimestamp((ms / 1000), tz=timezone.utc).strftime("%H:%M:%S.%f")[:-3], + ) + for i, ms in enumerate(x["breakOffset"] for x in track.data["adverts"]["breaks"]) + ] + + if track.data.get("endCredits", {}).get("squeezeIn"): + chapters.append( + Chapter( + name="Credits", + timestamp=datetime.fromtimestamp( + (track.data["endCredits"]["squeezeIn"] / 1000), tz=timezone.utc + ).strftime("%H:%M:%S.%f")[:-3], + ) + ) + + return chapters + + def get_widevine_service_certificate(self, **_: Any) -> str: + return WidevineCdm.common_privacy_cert + + def get_widevine_license(self, challenge: bytes, **_: Any) -> str: + payload = { + "message": base64.b64encode(challenge).decode("utf8"), + "token": self.license_token, + "request_id": self.asset_id, + "video": {"type": "ondemand", "url": self.manifest}, + } + + r = self.session.post(self.config["endpoints"]["license"], json=payload) + if not r.ok: + raise ConnectionError(f"License request failed: {r.json()['status']['type']}") + + return r.json()["license"] + + # Service specific functions + + def sort_assets(self, title: Union[Movie, Episode], android_assets: tuple, web_assets: tuple) -> tuple: + android_heights = None + web_heights = None + + if android_assets is not None: + try: + a_manifest, a_token, a_subtitle, data = android_assets + android_tracks = DASH.from_url(a_manifest, self.session).to_tracks(title.language) + android_heights = sorted([int(track.height) for track in android_tracks.videos], reverse=True) + except Exception: + android_heights = None + + if web_assets is not None: + try: + b_manifest, b_token, b_subtitle, data = web_assets + session = self.session + session.headers.update(self.config["headers"]) + web_tracks = DASH.from_url(b_manifest, session).to_tracks(title.language) + web_heights = sorted([int(track.height) for track in web_tracks.videos], reverse=True) + except Exception: + web_heights = None + + if not android_heights and not web_heights: + self.log.error("Failed to request manifest data. If you're behind a VPN/proxy, you might be blocked") + sys.exit(1) + + if not android_heights or android_heights[0] < 1080: + lic_token = self.decrypt_token(b_token, client="WEB") + return b_manifest, lic_token, b_subtitle, data + else: + lic_token = self.decrypt_token(a_token, client="ANDROID") + return a_manifest, lic_token, a_subtitle, data + + def android_playlist(self, video_id: str) -> tuple: + url = self.config["android"]["vod"].format(video_id=video_id) + headers = {"authorization": self.authorization} + + r = self.session.get(url=url, headers=headers) + if not r.ok: + self.log.warning("Request for Android endpoint returned %s", r) + return None + + data = json.loads(r.content) + manifest = data["videoProfiles"][0]["streams"][0]["uri"] + token = data["videoProfiles"][0]["streams"][0]["token"] + subtitle = next( + (x["url"] for x in data["subtitlesAssets"] if x["url"].endswith(".vtt")), + None, + ) + + return manifest, token, subtitle, data + + def web_playlist(self, video_id: str) -> tuple: + url = self.config["web"]["vod"].format(programmeId=video_id) + r = self.session.get(url, headers=self.config["headers"]) + if not r.ok: + self.log.warning("Request for WEB endpoint returned %s", r) + return None + + data = json.loads(r.content) + + for item in data["videoProfiles"]: + if item["name"] == "dashwv-dyn-stream-1": + token = item["streams"][0]["token"] + manifest = item["streams"][0]["uri"] + + subtitle = next( + (x["url"] for x in data["subtitlesAssets"] if x["url"].endswith(".vtt")), + None, + ) + + return manifest, token, subtitle, data + + def decrypt_token(self, token: str, client: str) -> tuple: + if client == "ANDROID": + key = self.config["android"]["key"] + iv = self.config["android"]["iv"] + + if client == "WEB": + key = self.config["web"]["key"] + iv = self.config["web"]["iv"] + + if isinstance(token, str): + token = base64.b64decode(token) + cipher = AES.new( + key=base64.b64decode(key), + iv=base64.b64decode(iv), + mode=AES.MODE_CBC, + ) + data = unpad(cipher.decrypt(token), AES.block_size) + dec_token = data.decode().split("|")[1] + return dec_token.strip() + + def get_html(self, url: str) -> dict: + r = self.session.get(url=url, headers=self.config["headers"]) + r.raise_for_status() + + init_data = re.search( + "", + "".join(r.content.decode().replace("\u200c", "").replace("\r\n", "").replace("undefined", "null")), + ) + try: + data = json.loads(init_data.group(1)) + return data["initialData"] + except Exception: + self.log.error(f"Failed to get episode for {url}") + sys.exit(1) diff --git a/packages/envied/src/envied/services/services/ALL4/config.yaml b/packages/envied/src/envied/services/services/ALL4/config.yaml new file mode 100644 index 0000000..1468e01 --- /dev/null +++ b/packages/envied/src/envied/services/services/ALL4/config.yaml @@ -0,0 +1,27 @@ +headers: + Accept-Language: en-US,en;q=0.8 + User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/77.0.3865.75 Safari/537.36 + +endpoints: + login: https://api.channel4.com/online/v2/auth/token + title: https://api.channel4.com/online/v1/views/content-hubs/{title}.json + license: https://c4.eme.lp.aws.redbeemedia.com/wvlicenceproxy-service/widevine/acquire + search: https://all4nav.channel4.com/v1/api/search + +android: + key: QVlESUQ4U0RGQlA0TThESA==" + iv: MURDRDAzODNES0RGU0w4Mg==" + auth: MzZVVUN0OThWTVF2QkFnUTI3QXU4ekdIbDMxTjlMUTE6Sllzd3lIdkdlNjJWbGlrVw== + vod: https://api.channel4.com/online/v1/vod/stream/{video_id}?client=android-mod + +web: + key: bjljTGllWWtxd3pOQ3F2aQ== + iv: b2R6Y1UzV2RVaVhMdWNWZA== + vod: https://www.channel4.com/vod/stream/{programmeId} + +device: + platform_name: android + device_type: mobile + device_name: "Sony C6903 (C6903)" + app_version: "android_app:9.4.2" + optimizely_datafile: "2908" diff --git a/packages/envied/src/envied/services/services/ARD/__init__.py b/packages/envied/src/envied/services/services/ARD/__init__.py new file mode 100644 index 0000000..a59587c --- /dev/null +++ b/packages/envied/src/envied/services/services/ARD/__init__.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +from http.cookiejar import MozillaCookieJar +from typing import Any, Optional, Union +from functools import partial +from pathlib import Path +import sys +import re + +import click +import webvtt +import requests +from click import Context +from bs4 import BeautifulSoup + +from envied.core.credential import Credential +from envied.core.service import Service +from envied.core.titles import Movie, Movies, Episode, Series +from envied.core.tracks import Track, Chapter, Tracks, Video, Subtitle +from envied.core.manifests.hls import HLS +from envied.core.manifests.dash import DASH + + +class ARD(Service): + """ + Service code for ARD Mediathek (https://www.ardmediathek.de) + + \b + Version: 1.0.0 + Author: lambda + Authorization: None + Robustness: + Unencrypted: 2160p, AAC2.0 + """ + + GEOFENCE = ("de",) + TITLE_RE = r"^(https://www\.ardmediathek\.de/(?Pserie|video)/.+/)(?P[a-zA-Z0-9]{10,})(/[0-9]{1,3})?$" + EPISODE_NAME_RE = r"^(Folge [0-9]+:)?(?P[^\(]+) \(S(?P[0-9]+)/E(?P[0-9]+)\)$" + + @staticmethod + @click.command(name="ARD", short_help="https://www.ardmediathek.de", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> ARD: + return ARD(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: + pass + + def get_titles(self) -> Union[Movies, Series]: + match = re.match(self.TITLE_RE, self.title) + if not match: + return + + item_id = match.group("item_id") + if match.group("item_type") == "video": + return self.load_player(item_id) + + r = self.session.get(self.config["endpoints"]["grouping"].format(item_id=item_id)) + item = r.json() + + for widget in item["widgets"]: + if widget["type"] == "gridlist" and widget.get("compilationType") == "itemsOfShow": + episodes = Series() + for teaser in widget["teasers"]: + if teaser["coreAssetType"] != "EPISODE": + continue + + if 'Hörfassung' in teaser['longTitle']: + continue + + episodes += self.load_player(teaser["id"]) + return episodes + + def get_tracks(self, title: Union[Episode, Movie]) -> Tracks: + if title.data["blockedByFsk"]: + self.log.error( + "This content is age-restricted and not currently available. " + "Try again after 10pm German time") + sys.exit(0) + + media_collection = title.data["mediaCollection"]["embedded"] + tracks = Tracks() + for stream_collection in media_collection["streams"]: + if stream_collection["kind"] != "main": + continue + + for stream in stream_collection["media"]: + if stream["mimeType"] == "application/vnd.apple.mpegurl": + tracks += Tracks(HLS.from_url(stream["url"]).to_tracks(stream["audios"][0]["languageCode"])) + break + + # Fetch tracks from HBBTV endpoint to check for potential H.265/2160p DASH + r = self.session.get(self.config["endpoints"]["hbbtv"].format(item_id=title.id)) + hbbtv = r.json() + for stream in hbbtv["video"]["streams"]: + for media in stream["media"]: + if media["mimeType"] == "application/dash+xml" and media["audios"][0]["kind"] == "standard": + tracks += Tracks(DASH.from_url(media["url"]).to_tracks(media["audios"][0]["languageCode"])) + break + + # for stream in title.data["video"]["streams"]: + # for media in stream["media"]: + # if media["mimeType"] != "video/mp4" or media["audios"][0]["kind"] != "standard": + # continue + + # tracks += Video( + # codec=Video.Codec.AVC, # Should check media["videoCodec"] + # range_=Video.Range.SDR, # Should check media["isHighDynamicRange"] + # width=media["maxHResolutionPx"], + # height=media["maxVResolutionPx"], + # url=media["url"], + # language=media["audios"][0]["languageCode"], + # fps=50, + # ) + + for sub in media_collection["subtitles"]: + for source in sub["sources"]: + if source["kind"] == "ebutt": + tracks.add(Subtitle( + codec=Subtitle.Codec.TimedTextMarkupLang, + language=sub["languageCode"], + url=source["url"] + )) + + return tracks + + def get_chapters(self, title: Union[Episode, Movie]) -> list[Chapter]: + return [] + + def load_player(self, item_id): + r = self.session.get(self.config["endpoints"]["item"].format(item_id=item_id)) + item = r.json() + + for widget in item["widgets"]: + if widget["type"] != "player_ondemand": + continue + + common_data = { + "id_": item_id, + "data": widget, + "service": self.__class__, + "language": "de", + "year": widget["broadcastedOn"][0:4], + } + + if widget["show"]["coreAssetType"] == "SINGLE" or not widget["show"].get("availableSeasons"): + return Movies([Movie( + name=widget["title"], + **common_data + )]) + else: + match = re.match(self.EPISODE_NAME_RE, widget["title"]) + if not match: + name = widget["title"] + season = 0 + episode = 0 + else: + name = match.group("name") + season = match.group("season") or 0 + episode = match.group("episode") or 0 + + return Series([Episode( + name=name, + title=widget["show"]["title"], + #season=widget["show"]["availableSeasons"][0], + season=season, + number=episode, + **common_data + )]) + diff --git a/packages/envied/src/envied/services/services/ARD/config.yaml b/packages/envied/src/envied/services/services/ARD/config.yaml new file mode 100644 index 0000000..e54c434 --- /dev/null +++ b/packages/envied/src/envied/services/services/ARD/config.yaml @@ -0,0 +1,8 @@ +headers: + Accept-Language: de-DE,de;q=0.8 + User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0 + +endpoints: + item: https://api.ardmediathek.de/page-gateway/pages/ard/item/{item_id}?embedded=true&mcV6=true + grouping: https://api.ardmediathek.de/page-gateway/pages/ard/grouping/{item_id}?seasoned=true&embedded=true + hbbtv: https://tv.ardmediathek.de/dyn/get?id=video:{item_id} diff --git a/packages/envied/src/envied/services/services/AUBC/__init__.py b/packages/envied/src/envied/services/services/AUBC/__init__.py new file mode 100644 index 0000000..6c4c1ad --- /dev/null +++ b/packages/envied/src/envied/services/services/AUBC/__init__.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Generator +from typing import Any, Optional, Union +from urllib.parse import urljoin + +import click +from click import Context +from requests import Request +from envied.core.constants import AnyTrack +from envied.core.manifests.dash import DASH +from envied.core.search_result import SearchResult +from envied.core.service import Service +from envied.core.titles import Episode, Movie, Movies, Series +from envied.core.tracks import Chapter, Chapters, Subtitle, Tracks + + +class AUBC(Service): + """ + \b + Service code for ABC iView streaming service (https://iview.abc.net.au/). + + \b + Version: 1.0.2 + Author: stabbedbybrick + Authorization: None + Robustness: + L3: 1080p, AAC2.0 + + \b + Tips: + - Input should be complete URL: + SHOW: https://iview.abc.net.au/show/return-to-paradise + EPISODE: https://iview.abc.net.au/video/DR2314H001S00 + MOVIE: https://iview.abc.net.au/show/way-back / https://iview.abc.net.au/show/way-back/video/ZW3981A001S00 + + """ + + GEOFENCE = ("au",) + ALIASES = ("iview", "abciview", "iv",) + + @staticmethod + @click.command(name="AUBC", short_help="https://iview.abc.net.au/", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> AUBC: + return AUBC(ctx, **kwargs) + + def __init__(self, ctx: Context, title: str): + self.title = title + super().__init__(ctx) + + self.session.headers.update(self.config["headers"]) + + def search(self) -> Generator[SearchResult, None, None]: + url = ( + "https://y63q32nvdl-1.algolianet.com/1/indexes/*/queries?x-algolia-agent=Algolia" + "%20for%20JavaScript%20(4.9.1)%3B%20Browser%20(lite)%3B%20react%20(17.0.2)%3B%20" + "react-instantsearch%20(6.30.2)%3B%20JS%20Helper%20(3.10.0)&x-" + "algolia-api-key=bcdf11ba901b780dc3c0a3ca677fbefc&x-algolia-application-id=Y63Q32NVDL" + ) + payload = { + "requests": [ + { + "indexName": "ABC_production_iview_web", + "params": f"query={self.title}&tagFilters=&userToken=anonymous-74be3cf1-1dc7-4fa1-9cff-19592162db1c", + } + ], + } + + results = self._request("POST", url, payload=payload)["results"] + hits = [x for x in results[0]["hits"] if x["docType"] == "Program"] + + for result in hits: + yield SearchResult( + id_="https://iview.abc.net.au/show/{}".format(result.get("slug")), + title=result.get("title"), + description=result.get("synopsis"), + label=result.get("subType"), + url="https://iview.abc.net.au/show/{}".format(result.get("slug")), + ) + + def get_titles(self) -> Union[Movies, Series]: + title_re = r"^(?:https?://(?:www.)?iview.abc.net.au/(?Pshow|video)/)?(?P[a-zA-Z0-9_-]+)" + try: + kind, title_id = (re.match(title_re, self.title).group(i) for i in ("type", "id")) + except Exception: + raise ValueError("- Could not parse ID from title") + + if kind == "show": + data = self._request("GET", "/v3/show/{}".format(title_id)) + label = data.get("type") + + if label.lower() in ("series", "program"): + episodes = self._series(title_id) + return Series(episodes) + + elif label.lower() in ("feature", "movie"): + movie = self._movie(data) + return Movies(movie) + + elif kind == "video": + episode = self._episode(title_id) + return Series([episode]) + + def get_tracks(self, title: Union[Movie, Episode]) -> Tracks: + video = self._request("GET", "/v3/video/{}".format(title.id)) + if not video.get("playable"): + raise ConnectionError(video.get("unavailableMessage")) + + playlist = video.get("_embedded", {}).get("playlist", {}) + if not playlist: + raise ConnectionError("Could not find a playlist for this title") + + streams = next(x["streams"]["mpegdash"] for x in playlist if x["type"] == "program") + captions = next((x.get("captions") for x in playlist if x["type"] == "program"), None) + title.data["protected"] = streams.get("protected", False) + + if "720" in streams: + streams["1080"] = streams["720"].replace("720", "1080") + + manifest = next( + (url for key in ["1080", "720", "sd", "sd-low"] if key in streams + for url in [streams[key]] + if self.session.head(url).status_code == 200), + None + ) + if not manifest: + raise ValueError("Could not find a manifest for this title") + + tracks = DASH.from_url(manifest, self.session).to_tracks(title.language) + + for track in tracks.audio: + role = track.data["dash"]["adaptation_set"].find("Role") + if role is not None and role.get("value") in ["description", "alternative", "alternate"]: + track.descriptive = True + + if captions: + subtitles = captions.get("src-vtt") + tracks.add( + Subtitle( + id_=hashlib.md5(subtitles.encode()).hexdigest()[0:6], + url=subtitles, + codec=Subtitle.Codec.from_mime(subtitles[-3:]), + language=title.language, + forced=False, + ) + ) + + return tracks + + def get_chapters(self, title: Union[Movie, Episode]) -> Chapters: + if not title.data.get("cuePoints"): + return Chapters() + + credits = next((x.get("start") for x in title.data["cuePoints"] if x["type"] == "end-credits"), None) + if credits: + return Chapters([Chapter(name="Credits", timestamp=credits * 1000)]) + + return 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 title.data.get("protected"): + return None + + customdata = self._license(title.id) + headers = {"customdata": customdata} + + r = self.session.post(self.config["endpoints"]["license"], headers=headers, data=challenge) + r.raise_for_status() + return r.content + + # Service specific + + def _series(self, title: str) -> Episode: + data = self._request("GET", "/v3/series/{}".format(title)) + + episodes = [ + self.create_episode(episode) + for season in data + for episode in reversed(season["_embedded"]["videoEpisodes"]["items"]) + if season.get("episodeCount") + ] + return Series(episodes) + + def _movie(self, data: dict) -> Movie: + return [ + Movie( + id_=data["_embedded"]["highlightVideo"]["id"], + service=self.__class__, + name=data.get("title"), + year=data.get("productionYear"), + data=data, + language=data.get("analytics", {}).get("dataLayer", {}).get("d_language", "en"), + ) + ] + + def _episode(self, video_id: str) -> Episode: + data = self._request("GET", "/v3/video/{}".format(video_id)) + return self.create_episode(data) + + def _license(self, video_id: str): + token = self._request("POST", "/v3/token/jwt", data={"clientId": self.config["client"]})["token"] + response = self._request("GET", "/v3/token/drm/{}".format(video_id), headers={"bearer": token}) + + return response["license"] + + def create_episode(self, episode: dict) -> Episode: + title = episode["showTitle"] + series_id = episode.get("analytics", {}).get("dataLayer", {}).get("d_series_id", "") + episode_name = episode.get("analytics", {}).get("dataLayer", {}).get("d_episode_name", "") + number = re.search(r"Episode (\d+)", episode.get("displaySubtitle", "")) + name = re.search(r"S\d+\sEpisode\s\d+\s(.*)", episode_name) + + language = episode.get("analytics", {}).get("dataLayer", {}).get("d_language", "en") + + return Episode( + id_=episode["id"], + service=self.__class__, + title=title, + season=int(series_id.split("-")[-1]) if series_id else 0, + number=int(number.group(1)) if number else 0, + name=name.group(1) if name else None, + data=episode, + language=language, + ) + + def _request(self, method: str, api: str, **kwargs: Any) -> Any[dict | str]: + url = urljoin(self.config["endpoints"]["base_url"], api) + + prep = self.session.prepare_request(Request(method, url, **kwargs)) + + response = self.session.send(prep) + if response.status_code != 200: + raise ConnectionError(f"{response.text}") + + try: + return json.loads(response.content) + + except json.JSONDecodeError as e: + raise ValueError(f"Failed to parse JSON: {response.text}") from e + diff --git a/packages/envied/src/envied/services/services/AUBC/config.yaml b/packages/envied/src/envied/services/services/AUBC/config.yaml new file mode 100644 index 0000000..604b739 --- /dev/null +++ b/packages/envied/src/envied/services/services/AUBC/config.yaml @@ -0,0 +1,9 @@ +headers: + User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:126.0) Gecko/20100101 Firefox/126.0 + accept-language: en-US,en;q=0.8 + +endpoints: + base_url: https://api.iview.abc.net.au + license: https://wv-keyos.licensekeyserver.com/ + +client: "1d4b5cba-42d2-403e-80e7-34565cdf772d" \ No newline at end of file diff --git a/packages/envied/src/envied/services/services/CBC/__init__.py b/packages/envied/src/envied/services/services/CBC/__init__.py new file mode 100644 index 0000000..d84ac0b --- /dev/null +++ b/packages/envied/src/envied/services/services/CBC/__init__.py @@ -0,0 +1,319 @@ +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 requests import Request +from envied.core.constants import AnyTrack +from envied.core.credential import Credential +from envied.core.manifests import DASH, HLS +from envied.core.search_result import SearchResult +from envied.core.service import Service +from envied.core.titles import Episode, Movie, Movies, Series +from envied.core.tracks import Chapter, Chapters, Tracks + + +class CBC(Service): + """ + \b + Service code for CBC Gem streaming service (https://gem.cbc.ca/). + + \b + Version: 1.0.1 + Author: stabbedbybrick + Authorization: Credentials + Robustness: + AES-128: 1080p, DDP5.1 + Widevine L3: 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: + - DRM encrypted titles max out at 720p. + - CCExtrator v0.94 will likely fail to extract subtitles. It's recommended to downgrade to v0.93. + - 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: str = title + super().__init__(ctx) + + self.base_url: str = self.config["endpoints"]["base_url"] + + def search(self) -> Generator[SearchResult, None, None]: + params = { + "device": "web", + "pageNumber": "1", + "pageSize": "20", + "term": self.title, + } + response: dict = 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.") + + tokens: Optional[Any] = self.cache.get(f"tokens_{credential.sha1}") + + """ + All grant types for future reference: + PASSWORD("password"), + ACCESS_TOKEN("access_token"), + REFRESH_TOKEN("refresh_token"), + CLIENT_CREDENTIALS("client_credentials"), + AUTHORIZATION_CODE("authorization_code"), + CODE("code"); + """ + + if tokens and not tokens.expired: + # cached + self.log.info(" + Using cached tokens") + auth_token: str = tokens.data["access_token"] + + elif tokens and tokens.expired: + # expired, refresh + self.log.info("Refreshing cached tokens...") + auth_url, scopes = self.settings() + params = { + "client_id": self.config["client"]["id"], + "grant_type": "refresh_token", + "refresh_token": tokens.data["refresh_token"], + "scope": scopes, + } + + access: dict = self._request("POST", auth_url, params=params) + + # Shorten expiration by one hour to account for clock skew + tokens.set(access, expiration=int(access["expires_in"]) - 3600) + auth_token: str = access["access_token"] + + else: + # new + self.log.info("Requesting new tokens...") + auth_url, scopes = self.settings() + params = { + "client_id": self.config["client"]["id"], + "grant_type": "password", + "username": credential.username, + "password": credential.password, + "scope": scopes, + } + + access: dict = self._request("POST", auth_url, params=params) + + # Shorten expiration by one hour to account for clock skew + tokens.set(access, expiration=int(access["expires_in"]) - 3600) + auth_token: str = access["access_token"] + + claims_token: str = self.claims_token(auth_token) + self.session.headers.update({"x-claims-token": claims_token}) + + def get_titles(self) -> Union[Movies, Series]: + title_re: str = r"^(?:https?://(?:www.)?gem.cbc.ca/)?(?P[a-zA-Z0-9_-]+)" + try: + title_id: str = re.match(title_re, self.title).group("id") + except Exception: + raise ValueError("- Could not parse ID from title") + + params = {"device": "web"} + data: dict = self._request("GET", "/ott/catalog/v2/gem/show/{}".format(title_id), params=params) + label: str = data.get("contentType", "").lower() + + if label in ("film", "movie", "standalone"): + movies: list[Movie] = self._movie(data) + return Movies(movies) + + else: + episodes: list[Episode] = self._show(data) + return Series(episodes) + + def get_tracks(self, title: Union[Movie, Episode]) -> Tracks: + index: dict = self._request( + "GET", "/media/meta/v1/index.ashx", params={"appCode": "gem", "idMedia": title.id, "output": "jsonObject"} + ) + + title.data["extra"] = { + "chapters": index["Metas"].get("Chapitres"), + "credits": index["Metas"].get("CreditStartTime"), + } + + self.drm: bool = index["Metas"].get("isDrmActive") == "true" + if self.drm: + tech: str = next(tech["name"] for tech in index["availableTechs"] if "widevine" in tech["drm"]) + else: + tech: str = next(tech["name"] for tech in index["availableTechs"] if not tech["drm"]) + + response: dict = self._request( + "GET", self.config["endpoints"]["validation"].format("android", title.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: Union[HLS, DASH] = HLS if tech == "hls" else DASH + tracks: Tracks = stream_type.from_url(manifest, self.session).to_tracks(language=title.language) + + 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 + + for track in tracks: + track.language = title.language + + return tracks + + def get_chapters(self, title: Union[Movie, Episode]) -> Chapters: + extra: dict = title.data["extra"] + + chapters = [] + if extra.get("chapters"): + chapters = [Chapter(timestamp=x) for x in set(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) -> list[Episode]: + lineups: list = next((x["lineups"] for x in data["content"] if x.get("title", "").lower() == "episodes"), None) + if not lineups: + self.log.warning("No episodes found for: {}".format(data.get("title"))) + return + + titles = [] + for season in lineups: + for episode in season["items"]: + if episode.get("mediaType", "").lower() == "episode": + parts = episode.get("title", "").split(".", 1) + episode_name = parts[1].strip() if len(parts) > 1 else parts[0].strip() + titles.append( + Episode( + id_=episode["idMedia"], + service=self.__class__, + title=data.get("title"), + season=int(season.get("seasonNumber", 0)), + number=int(episode.get("episodeNumber", 0)), + name=episode_name, + year=episode.get("metadata", {}).get("productionYear"), + language=data["structuredMetadata"].get("inLanguage", "en-CA"), + data=episode, + ) + ) + + return titles + + def _movie(self, data: dict) -> list[Movie]: + unwanted: tuple = ("episodes", "trailers", "extras") + lineups: list = next((x["lineups"] for x in data["content"] if x.get("title", "").lower() not in unwanted), None) + if not lineups: + self.log.warning("No movies found for: {}".format(data.get("title"))) + return + + titles = [] + for season in lineups: + for movie in season["items"]: + if movie.get("mediaType", "").lower() == "episode": + parts = movie.get("title", "").split(".", 1) + movie_name = parts[1].strip() if len(parts) > 1 else parts[0].strip() + titles.append( + Movie( + id_=movie.get("idMedia"), + service=self.__class__, + name=movie_name, + year=movie.get("metadata", {}).get("productionYear"), + language=data["structuredMetadata"].get("inLanguage", "en-CA"), + data=movie, + ) + ) + + return titles + + def settings(self) -> tuple: + settings = self._request("GET", "/ott/catalog/v1/gem/settings", params={"device": "web"}) + auth_url: str = settings["identityManagement"]["ropc"]["url"] + scopes: str = settings["identityManagement"]["ropc"]["scopes"] + return auth_url, scopes + + def claims_token(self, token: str) -> str: + headers = { + "Authorization": "Bearer " + token, + } + params = {"device": "web"} + response: dict = self._request( + "GET", "/ott/subscription/v2/gem/Subscriber/profile", headers=headers, params=params + ) + return response["claimsToken"] + + def _request(self, method: str, api: str, **kwargs: Any) -> Any[dict | str]: + url: str = urljoin(self.base_url, api) + + prep: Request = self.session.prepare_request(Request(method, url, **kwargs)) + 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)) diff --git a/packages/envied/src/envied/services/services/CBC/config.yaml b/packages/envied/src/envied/services/services/CBC/config.yaml new file mode 100644 index 0000000..31bd19a --- /dev/null +++ b/packages/envied/src/envied/services/services/CBC/config.yaml @@ -0,0 +1,7 @@ +endpoints: + base_url: "https://services.radio-canada.ca" + validation: "/media/validation/v2?appCode=gem&&deviceType={}&idMedia={}&manifestType={}&output=json&tech={}" + api_key: "3f4beddd-2061-49b0-ae80-6f1f2ed65b37" + +client: + id: "fc05b0ee-3865-4400-a3cc-3da82c330c23" \ No newline at end of file diff --git a/packages/envied/src/envied/services/services/CBS/__init__.py b/packages/envied/src/envied/services/services/CBS/__init__.py new file mode 100644 index 0000000..ae6792d --- /dev/null +++ b/packages/envied/src/envied/services/services/CBS/__init__.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +import json +import re +import sys +from collections.abc import Generator +from typing import Any, Optional, Union +from urllib.parse import urljoin + +import click +from requests import Request +from envied.core.constants import AnyTrack +from envied.core.manifests import DASH +from envied.core.search_result import SearchResult +from envied.core.service import Service +from envied.core.titles import Episode, Series, Title_T, Titles_T +from envied.core.tracks import Chapter, Chapters, Tracks +from envied.core.utils.sslciphers import SSLCiphers +from envied.core.utils.xml import load_xml + + +class CBS(Service): + """ + \b + Service code for CBS.com streaming service (https://cbs.com). + Credit to @srpen6 for the tip on anonymous session + + \b + Version: 1.0.1 + Author: stabbedbybrick + Authorization: None + Robustness: + Widevine: + L3: 2160p, DDP5.1 + + \b + Tips: + - Input should be complete URLs: + SERIES: https://www.cbs.com/shows/tracker/ + EPISODE: https://www.cbs.com/shows/video/E0wG_ovVMkLlHOzv7KDpUV9bjeKFFG2v/ + + \b + Common VPN/proxy errors: + - SSLError(SSLEOFError(8, '[SSL: UNEXPECTED_EOF_WHILE_READING]')) + - ConnectionError: 406 Not Acceptable, 403 Forbidden + + """ + + GEOFENCE = ("us",) + + @staticmethod + @click.command(name="CBS", short_help="https://cbs.com", help=__doc__) + @click.argument("title", type=str, required=False) + @click.pass_context + def cli(ctx, **kwargs) -> CBS: + return CBS(ctx, **kwargs) + + def __init__(self, ctx, title): + self.title = title + super().__init__(ctx) + + 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 get_titles(self) -> Titles_T: + title_re = r"https://www\.cbs\.com/shows/(?P