diff --git a/packages/envied/src/envied/core/drm/widevine.py b/packages/envied/src/envied/core/drm/widevine.py index f4d7a82..008e9d9 100644 --- a/packages/envied/src/envied/core/drm/widevine.py +++ b/packages/envied/src/envied/core/drm/widevine.py @@ -348,7 +348,7 @@ class Widevine: try: arguments = [ - f"input={path},stream=0,output={output_path},output_format=MP4", + f"input={path},stream=0,output={output_path},output_format=mp4", "--enable_raw_key_decryption", "--keys", ",".join( diff --git a/packages/envied/src/envied/core/tracks/subtitle.py b/packages/envied/src/envied/core/tracks/subtitle.py index 7135692..4b625a6 100644 --- a/packages/envied/src/envied/core/tracks/subtitle.py +++ b/packages/envied/src/envied/core/tracks/subtitle.py @@ -861,13 +861,20 @@ class Subtitle(Track): Subtitle.Codec.SAMI: "sami", Subtitle.Codec.MicroDVD: "microdvd", }.get(codec, codec.name.lower()) - sub_edit_args = [ + '''sub_edit_args = [ str(binaries.SubtitleEdit), "/convert", str(self.path), sub_edit_format, f"/outputfilename:{output_path.name}", "/encoding:utf8", + ]''' + ## Angela edited this to remove the /convert argument because it was causing issues with SubtitleEdit + sub_edit_args = [ + binaries.SubtitleEdit, + self.path, sub_edit_format, + f"/outputfilename:{output_path.name}", + "/encoding:utf8" ] if codec == Subtitle.Codec.SubRip: sub_edit_args.append("/ConvertColorsToDialog") diff --git a/packages/envied/src/envied/envied.yaml.save b/packages/envied/src/envied/envied.yaml.save new file mode 100644 index 0000000..9ef8af3 --- /dev/null +++ b/packages/envied/src/envied/envied.yaml.save @@ -0,0 +1,274 @@ +# Group or Username to postfix to the end of all download fienames following a dash +#tag: '' + +# Set terminal background color (custom option not in CONFIG.md) +set_terminal_bg: false + +# Muxing configuration +muxing: + set_title: false + +# shakapackager or mp4decrypt +decryption: shakapackager + +# Custom output templates for filenames +# Configure output_template in your unshackle.yaml to control filename format. +# If not configured, default scene-style templates are used and a warning is shown. +# Available variables: {title}, {year}, {season}, {episode}, {season_episode}, {episode_name}, +# {quality}, {resolution}, {source}, {audio}, {audio_channels}, {audio_full}, +# {video}, {hdr}, {hfr}, {atmos}, {dual}, {multi}, {tag}, {edition}, {repack}, +# {lang_tag} +# Conditional variables (included only if present): Add ? suffix like {year?}, {episode_name?}, {hdr?} +# Customize the templates below: +# +# Example outputs: +# Scene movies: 'The.Matrix.1999.1080p.SERVICE.WEB-DL.DDP5.1.H.264-EXAMPLE' +# Scene movies (HDR): 'Dune.2021.2160p.SERVICE.WEB-DL.DDP5.1.HDR10.H.265-EXAMPLE' +# Scene movies (REPACK): 'Dune.2021.REPACK.2160p.SERVICE.WEB-DL.DDP5.1.H.265-EXAMPLE' +# Scene series: 'Breaking.Bad.2008.S01E01.Pilot.1080p.SERVICE.WEB-DL.DDP5.1.H.264-EXAMPLE' +# Plex movies: 'The Matrix (1999)' +# Plex series: 'Breaking Bad S01E01 Pilot' +output_template: + # Scene-style naming (dot-separated) + #movies: '{title}.{year}.{repack?}.{edition?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}' + #series: '{title}.{year?}.{season_episode}.{episode_name?}.{repack?}.{edition?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}' + #songs: '{track_number}.{title}.{repack?}.{edition?}.{source?}.WEB-DL.{audio_full}.{atmos?}-{tag}' + # + # Plex-friendly naming (space-separated, clean format) + movies: '{title} ({year}) {quality}' + series: '{title} {season_episode} {episode_name?}' + songs: '{track_number}. {title}' + # + # Minimal naming (basic info only) + # movies: '{title}.{year}.{quality}' + # series: '{title}.{season_episode}.{episode_name?}' + # + # Custom scene-style with specific elements + # movies: '{title}.{year}.{quality}.{hdr?}.{source}.WEB-DL.{audio_full}.{video}-{tag}' + # series: '{title}.{year?}.{season_episode}.{episode_name?}.{quality}.{hdr?}.{source}.WEB-DL.{audio_full}.{atmos?}.{video}-{tag}' + +# Language-based tagging for output filenames +# Automatically adds language identifiers (e.g., DANiSH, NORDiC, DKsubs) based on +# audio and subtitle track languages. Rules are evaluated in order; first match wins. +# Use {lang_tag?} in your output_template to place the tag in the filename. +# +# Conditions (all conditions in a rule must match): +# audio: - any audio track matches this language +# subs_contain: - any subtitle matches this language +# subs_contain_all: [lang, ...] - subtitles include ALL listed languages +# +# language_tags: +# rules: +# - audio: da +# tag: DANiSH +# - audio: sv +# tag: SWEDiSH +# - audio: nb +# tag: NORWEGiAN +# - audio: en +# subs_contain_all: [da, sv, nb] +# tag: NORDiC +# - audio: en +# subs_contain: da +# tag: DKsubs + + +# Widevine pssh display; fold, crop or ellipsis +pssh_display: ellipsis + +# Login credentials for each Service +credentials: + ALL4: angela.slaney@gmail.com:Nood13sh + ROKU: moretonplumber@yandex.com:$WPJ~Mdh2:X@EtB + TVNZ: moretonplumber@yandex.com:penumbra + TPTV: moretonplumber@yandex.com:penumbra1234 + CBC: moretonplumber@yandex.com:penumbra + CRAV: moretonplumber@yandex.com:Penumbra1$ + +# Override default directories used across unshackle +directories: + cache: Cache + cookies: Cookies + dcsl: DCSL # Device Certificate Status List + downloads: /home/angela/Downloads/devine/ + logs: Logs + temp: Temp + wvds: WVDs + prds: PRDs + # Additional directories that can be configured: + # commands: Commands + #services: + # - ./envied/services/ + vaults: vaults/ + +# fonts: Fonts + +# Pre-define which Widevine or PlayReady device to use for each Service +cdm: + default: device +# itv: DRMLABL3 + +# ========================================== +# REMOTE CDM - not working +# ========================================== + +remote_cdm: + - name: "DRMLABL3" + "Device Type": ANDROID + "Device Name": "l3" + "System ID": 8159 + "Security Level": 3 + "Host": "http://api2.drmlab.io" + "Secret": "drmlab" + +remote_cdm: + - name: "DRMLAB_SL3K" + "Device Type": PLAYREADY + "Device Name": "sl3k" + "System ID": 0 + "Security Level": 3000 + "Host": "http://api2.drmlab.io/playready" + "Secret": "drmlab" + +# Key Vaults store your obtained Content Encryption Keys (CEKs) +key_vaults: +- type: SQLite + name: Local vault + path: /home/angela/Programming/gits/devine-services/key_store.db + +#- type: HTTPAPI +# name: drmlab +# host: http://api.drmlab.io/vault/ +# password: gEX75q7I5YVkvgF5SUkcNd41IbGrDtTT +- type: HTTP + name: DRMLab + host: http://api.drmlab.io/vault/ + password: Qm9XvT4pZ8kLs2HdRw6YnBc7Ju3EfGa1 + api_mode: json + +- type: HTTP + name: Gang Slodziakow Vault + host: http://217.147.172.43/ + username: admin + password: 9BfsPtho4KD7LHpXhI3ZNvTR8p3TQb9iLWYtjRu366Nx9od0PAp8C37gPoDIDxl5 + +- type: API + name: LibreDRM Vault + uri: https://api.mapiro.com/vault + token: 5fc8fcead3bc4c1403dcec54fc3a326fbd3ec95da3899d24 + +- type: HTTP + name: Tupac Vault + host: https://keys.drm.hk + password: a8fGY4LObPLrt6hhT5J6pqAEqya5BJji6J4Y809gdcysW2Mg7Wo4QkrrVgecv0zD + api_mode: json + +# Choose what software to use to download data +#downloader: n_m3u8dl_re +# Options: requests | aria2c | curl_impersonate | n_m3u8dl_re +# Can also be a mapping: +downloader: +# NF: requests +# AMZN: n_m3u8dl_re + ALL4: n_m3u8dl_re + BBC: requests + default: n_m3u8dl_re + +# aria2c downloader configuration +aria2c: + max_concurrent_downloads: 4 + max_connection_per_server: 3 + split: 5 + file_allocation: falloc # none | prealloc | falloc | trunc + +# N_m3u8DL-RE downloader configuration +n_m3u8dl_re: + thread_count: 16 + ad_keyword: "advertisement" + use_proxy: true + +# curl_impersonate downloader configuration +curl_impersonate: + browser: chrome120 + +# Pre-define default options and switches of the dl command +dl: + best: true + sub_format: srt + downloads: 4 + workers: 16 + + + +# Chapter Name to use when exporting a Chapter without a Name +chapter_fallback_name: "Chapter {j:02}" + +# Case-Insensitive dictionary of headers for all Services +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" + +# Override default filenames used across unshackle +filenames: + log: "envied_{name}_{time}.log" + config: "config.yaml" + root_config: "envied.yaml" + chapters: "Chapters_{title}_{random}.txt" + subtitle: "Subtitle_{id}_{language}.srt" + +# API key for The Movie Database (TMDB) +#tmdb_api_key: "cc4e028567dd1bd804d650b891b86f6e" +tmdb_api_key: "502b935d7899aaaf64f0a1104b9bf488" +# tag_imdb_tmdb (bool) +tag_imdb_tmdb: true + + +#conversion_method: +# - auto (default): Smart routing - subby for WebVTT/SAMI, pycaption for others +# - subby: Always use subby with advanced processing +# - pycaption: Use only pycaption library (no SubtitleEdit, no subby) +# - subtitleedit: Prefer SubtitleEdit when available, fall back to pycaption +# - pysubs2: Use pysubs2 library (supports SRT/SSA/ASS/WebVTT/TTML/SAMI/MicroDVD/MPL2/TMP) +subtitle: + conversion_method: auto + # sdh_method: Method to use for SDH (hearing impaired) stripping + # - auto (default): Try subby (SRT only), then SubtitleEdit (if available), then subtitle-filter + # - subby: Use subby library (SRT only) + # - subtitleedit: Use SubtitleEdit tool (Windows only, falls back to subtitle-filter) + # - filter-subs: Use subtitle-filter library directly + sdh_method: auto + # strip_sdh: Automatically create stripped (non-SDH) versions of SDH subtitles + # Set to false to disable automatic SDH stripping entirely (default: true) + strip_sdh: true + # convert_before_strip: Auto-convert VTT/other formats to SRT before using subtitle-filter + # This ensures compatibility when subtitle-filter is used as fallback (default: true) + convert_before_strip: true + # preserve_formatting: Preserve original subtitle formatting (tags, positioning, styling) + # When true, skips pycaption processing for WebVTT files to keep tags like , , positioning intact + # Combined with no sub_format setting, ensures subtitles remain in their original format (default: true) + preserve_formatting: true + # output_mode: Output mode for subtitles + # - mux: Embed subtitles in MKV container only (default) + # - sidecar: Save subtitles as separate files only + # - both: Embed in MKV AND save as sidecar files + output_mode: mux + # sidecar_format: Format for sidecar subtitle files + # Options: srt, vtt, ass, original (keep current format) + sidecar_format: srt + + +# Configuration data for each Service +services: + # Service-specific configuration goes here + # EXAMPLE: + # api_key: "service_specific_key" + +# Legacy NordVPN configuration (use proxy_providers instead) +proxy_providers: + nordvpn: + username: gBM9ejjmY1J15jccH12VT7Qy + password: 1kmuCrRXnT55uAjiUTd8twv9 + server_map: + us: 6918 + uk: 2613 + nz: 100 diff --git a/packages/envied/src/envied/services/BLAZ/__init__.py b/packages/envied/src/envied/services/BLAZ/__init__.py new file mode 100644 index 0000000..5aa2dbd --- /dev/null +++ b/packages/envied/src/envied/services/BLAZ/__init__.py @@ -0,0 +1,387 @@ +from __future__ import annotations + +import html +import json +import re +from collections.abc import Generator +from typing import Any +from urllib.parse import urljoin, urlparse + +import click +from bs4 import BeautifulSoup +from click import Context +from requests import Request +from envied.core.manifests import HLS +from envied.core.search_result import SearchResult +from envied.core.service import Service +from envied.core.titles import Episode, Series, Title_T +from envied.core.tracks import Chapters, Tracks + + +class BLAZ(Service): + """ + \b + Service code for BLAZE TV's free streaming service (https://watch.blaze.tv/). + + \b + Version: 1.0.0 + Author: billybanana + Authorization: None + Geofence: GB/UK + Robustness: + HLS: 720p, AAC2.0 + + \b + Tips: + - Search is supported: + envied.dl BLAZ "outback wreckers" + - Show URLs, public Blaze URLs, slugs, and episode URLs are supported: + https://www.blaze.tv/outback-wreckers + https://watch.blaze.tv/shows/4090b77a-538c-11f1-b4ab-021644e4b9e7/outback-wreckers + outback-wreckers + https://watch.blaze.tv/watch/vod/53361682/island-dreams-or-french-connection + + """ + + ALIASES = ("blaze",) + GEOFENCE = ("GB", "UK") + TITLE_RE = ( + r"^(?:https?://)?" + r"(?:(?:www\.)?blaze\.tv/|watch\.blaze\.tv/shows/[a-f0-9-]+/)?" + r"(?P[a-z0-9-]+)$" + ) + EPISODE_RE = r"^(?:https?://watch\.blaze\.tv)?/watch/vod/(?P\d+)(?:/(?P[a-z0-9-]+))?" + + @staticmethod + @click.command(name="BLAZ", short_help="https://watch.blaze.tv/", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> BLAZ: + return BLAZ(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]: + query = self._clean_text(self.title).lower() + + for item in self._catalogue(): + haystack = f"{item['title']} {item['slug']}".lower() + if query and query not in haystack: + continue + + yield SearchResult( + id_=item["url"], + title=item["title"], + description=item.get("description"), + label="SERIES", + url=item["url"], + ) + + def get_titles(self) -> Series: + episode_match = re.match(self.EPISODE_RE, self.title) + if episode_match: + return Series(self._episode(episode_match.group("uvid"), self.title)) + + show_url = self._resolve_show_url(self.title) + html_text = self._request("GET", show_url) + return Series(self._series(show_url, html_text)) + + def get_tracks(self, title: Title_T) -> Tracks: + token = self._request("GET", self.config["endpoints"]["token"].format(uvid=title.id)) + if not token.get("token") or not token.get("expiry"): + raise ValueError(f"Failed to retrieve player token: {token}") + + data = title.data or {} + content_type = data.get("type", "replay").lower() + key = data.get("key") or self.config["playback"]["key"] + + stream = self._request( + "POST", + self.config["endpoints"]["streams"].format(type=content_type, uvid=title.id), + params={ + "key": key, + "platform": self.config["playback"]["platform"], + }, + headers={ + "Accept": "application/json", + "Origin": self.config["endpoints"]["base"], + "Referer": data.get("url") or self.config["endpoints"]["base"], + "Token": token["token"], + "Token-Expiry": str(token["expiry"]), + "Userid": "123456", + "Uvid": str(title.id), + }, + ) + + response = stream.get("response", {}) + if response.get("error"): + raise ValueError(f"Streams API error: {response['error']}") + manifest = response.get("stream") + if not manifest: + raise ValueError(f"Streams API did not return a manifest: {stream}") + + return HLS.from_url(manifest, self.session).to_tracks(language=title.language) + + def get_chapters(self, title: Title_T) -> Chapters: + return Chapters() + + def get_widevine_service_certificate(self, **_: Any) -> str | None: + return None + + def get_widevine_license(self, **_: Any) -> bytes | str | None: + return None + + # Service specific + + def _catalogue(self) -> list[dict[str, str]]: + html_text = self._request("GET", self.config["endpoints"]["series"]) + soup = BeautifulSoup(html_text, "html.parser") + by_url: dict[str, dict[str, str]] = {} + + for a in soup.find_all("a", href=True): + href = urljoin(self.config["endpoints"]["base"], a["href"]) + if "/shows/" not in urlparse(href).path: + continue + by_url[href] = { + "url": href, + "slug": self._slug_from_show_url(href), + "title": self._title_from_node(a) or self._title_from_slug(self._slug_from_show_url(href)), + "description": None, + } + + for match in re.finditer(r"https://watch\.blaze\.tv/shows/[^'\"<>\s]+", html_text): + href = html.unescape(match.group(0)) + if href in by_url: + continue + + snippet = html_text[max(0, match.start() - 1200): match.end() + 1200] + title = self._title_from_snippet(snippet) or self._title_from_slug(self._slug_from_show_url(href)) + by_url[href] = { + "url": href, + "slug": self._slug_from_show_url(href), + "title": title, + "description": None, + } + + return sorted(by_url.values(), key=lambda item: item["title"].lower()) + + def _resolve_show_url(self, title: str) -> str: + if re.match(r"^https?://watch\.blaze\.tv/shows/", title): + return title + + match = re.match(self.TITLE_RE, title) + if not match: + raise ValueError(f"Could not parse BLAZE title input: {title}") + + slug = match.group("slug") + for item in self._catalogue(): + if item["slug"] == slug: + return item["url"] + + raise ValueError(f"Could not resolve BLAZE show slug: {slug}") + + def _series(self, show_url: str, html_text: str) -> list[Episode]: + soup = BeautifulSoup(html_text, "html.parser") + show_title = self._show_title(soup) + + entries: list[dict[str, Any]] = [] + seen: set[str] = set() + for a in soup.find_all("a", href=True): + href = urljoin(show_url, a["href"]) + match = re.search(r"/watch/vod/(\d+)/([^/?#]+)", href) + if not match: + continue + + uvid, slug = match.groups() + if uvid in seen: + continue + seen.add(uvid) + + player = self._player_data(html_text, uvid) + entries.append( + { + "uvid": uvid, + "slug": slug, + "url": href, + "name": self._episode_title(a, slug), + "description": self._episode_description(a), + "player": player, + } + ) + + if not entries: + raise ValueError(f"No BLAZE VOD episodes found for {show_url}") + + total = len(entries) + episodes = [] + for index, item in enumerate(entries): + number = total - index + data = { + "url": item["url"], + "slug": item["slug"], + "type": item["player"].get("type", "replay"), + "key": item["player"].get("key") or self.config["playback"]["key"], + "poster": item["player"].get("poster"), + } + episodes.append( + Episode( + id_=item["uvid"], + service=self.__class__, + title=show_title, + season=1, + number=number, + name=item["name"], + language="en", + data=data, + description=item["description"], + ) + ) + + return episodes + + def _episode(self, uvid: str, episode_url: str) -> list[Episode]: + if episode_url.startswith("/"): + episode_url = urljoin(self.config["endpoints"]["base"], episode_url) + elif not episode_url.startswith("http"): + episode_url = f"{self.config['endpoints']['base']}/watch/vod/{uvid}/{episode_url}" + + html_text = self._request("GET", episode_url) + soup = BeautifulSoup(html_text, "html.parser") + player = self._player_data(html_text, uvid) + + return [ + Episode( + id_=uvid, + service=self.__class__, + title=self._show_title(soup), + season=1, + number=0, + name=self._page_title(soup) or self._title_from_slug(urlparse(episode_url).path.rstrip("/").split("/")[-1]), + language="en", + data={ + "url": episode_url, + "type": player.get("type", "replay"), + "key": player.get("key") or self.config["playback"]["key"], + "poster": player.get("poster"), + }, + ) + ] + + def _player_data(self, html_text: str, uvid: str) -> dict[str, str]: + soup = BeautifulSoup(html_text, "html.parser") + node = soup.find(attrs={"data-uvid": uvid}) + data: dict[str, str] = {} + if node: + for key, value in node.attrs.items(): + if key.startswith("data-"): + data[key[5:].replace("-", "_")] = str(value) + + data.setdefault("uvid", uvid) + data.setdefault("type", "replay") + data.setdefault("key", self.config["playback"]["key"]) + return data + + def _request(self, method: str, endpoint: str, **kwargs: Any) -> Any: + prep = self.session.prepare_request(Request(method, endpoint, **kwargs)) + response = self.session.send(prep) + + if response.status_code not in (200, 201): + raise ConnectionError(response.text) + + content_type = response.headers.get("content-type", "") + if "json" in content_type or response.text.strip().startswith(("{", "[")): + try: + return json.loads(response.content) + except json.JSONDecodeError as e: + raise ValueError(f"Failed to parse JSON: {response.text}") from e + + return response.text + + @staticmethod + def _clean_text(value: str | None) -> str: + return " ".join(html.unescape(value or "").split()) + + @classmethod + def _title_from_slug(cls, slug: str) -> str: + return cls._clean_text(slug.replace("-", " ").title()) + + @staticmethod + def _slug_from_show_url(url: str) -> str: + return urlparse(url).path.rstrip("/").split("/")[-1] + + @classmethod + def _title_from_node(cls, node: Any) -> str: + text = cls._clean_text(node.get_text(" ", strip=True)) + text = re.sub(r"\bPlay\b", "", text, flags=re.I) + text = re.sub(r"\b\d+\s+Episodes?\b", "", text, flags=re.I) + return cls._clean_text(text) + + @classmethod + def _title_from_snippet(cls, snippet: str) -> str | None: + alt = re.search(r'alt="([^"]+)"', snippet) + if alt: + title = cls._clean_text(alt.group(1)) + if title and title.lower() not in ("image", "menu"): + return title + + overlay = re.search(r'overlay-title[^>]*>\s*([^<]+)', snippet, re.I) + if overlay: + return cls._clean_text(overlay.group(1)) + + return None + + @classmethod + def _show_title(cls, soup: BeautifulSoup) -> str: + h1 = soup.find("h1") + if h1: + title = cls._clean_text(h1.get_text(" ", strip=True)) + if title and title.lower() not in ("series", "watch"): + return title + + og_title = soup.find("meta", property="og:title") + if og_title and og_title.get("content"): + title = re.sub(r"^BLAZE\s*::\s*", "", cls._clean_text(og_title["content"])).strip() + if title and title.lower() not in ("series", "watch"): + return title + + return "BLAZE" + + @classmethod + def _page_title(cls, soup: BeautifulSoup) -> str | None: + for selector in ({"property": "og:title"}, {"name": "twitter:title"}): + meta = soup.find("meta", attrs=selector) + if meta and meta.get("content"): + title = re.sub(r"^BLAZE\s*::\s*", "", cls._clean_text(meta["content"])).strip() + if title and title.lower() not in ("series", "watch"): + return title + + title_tag = soup.find("title") + if title_tag: + title = re.sub(r"^BLAZE\s*::\s*", "", cls._clean_text(title_tag.get_text())).strip() + if title and title.lower() not in ("series", "watch"): + return title + + return None + + @classmethod + def _episode_title(cls, link: Any, slug: str) -> str: + title = cls._title_from_node(link) + return title or cls._title_from_slug(slug) + + @classmethod + def _episode_description(cls, link: Any) -> str | None: + node = link + for _ in range(6): + if not node: + return None + text = cls._clean_text(node.get_text(" ", strip=True)) + if len(text) > 80: + title = cls._title_from_node(link) + text = cls._clean_text(text.replace(title, "", 1)) + return text or None + node = node.parent + return None diff --git a/packages/envied/src/envied/services/BLAZ/config.yaml b/packages/envied/src/envied/services/BLAZ/config.yaml new file mode 100644 index 0000000..501ecaf --- /dev/null +++ b/packages/envied/src/envied/services/BLAZ/config.yaml @@ -0,0 +1,13 @@ +headers: + User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36" + Accept: "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" + +endpoints: + base: "https://watch.blaze.tv" + series: "https://watch.blaze.tv/series" + token: "https://watch.blaze.tv/api/player/token/{uvid}" + streams: "https://v2-streams-elb.simplestreamcdn.com/api/{type}/stream/{uvid}" + +playback: + key: "6Pr3Uj2Fo3Ix7Rb9Ze9Ro4Ez2Eq7Sy" + platform: "chrome" diff --git a/packages/envied/src/envied/services/NINE/__init__.py b/packages/envied/src/envied/services/NINE/__init__.py new file mode 100644 index 0000000..db895e5 --- /dev/null +++ b/packages/envied/src/envied/services/NINE/__init__.py @@ -0,0 +1,517 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Generator +from datetime import datetime +from typing import Any +from urllib.parse import urljoin + +import click +from click import Context +from requests import Request +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, Subtitle, Tracks + + +class NINE(Service): + """ + Service code for 9Now streaming service (https://www.9now.com.au/). + + \b + Version: 1.0.0 + Authored: billybanana + Authorization: None + Geofence: AU (API and downloads) + Robustness: + Widevine: L3 720p, some 1080p clips + Clear HLS: up to source quality + + \b + Tips: + - Search by show name: + envied.dl NINE "Travel Guides" + - Use complete URLs: + SERIES: https://www.9now.com.au/travel-guides + EPISODE: https://www.9now.com.au/travel-guides/season-9/episode-2 + CLIP: https://www.9now.com.au/premier-league-epl-football/season-20252026/clip-cmeop4x67000m0hmmc1822v1i + """ + + GEOFENCE = ("au",) + ALIASES = ("9now", "nine",) + + @staticmethod + @click.command(name="NINE", short_help="https://www.9now.com.au/", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> NINE: + return NINE(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]: + results = self._request( + "GET", + self.config["endpoints"]["search"], + params={"q": self.title.strip(), "device": "web"}, + ) + + seen = set() + for group in results.get("results", []): + if group.get("title") != "Search results": + continue + + for result in group.get("items", []): + link = result.get("link") or {} + web_url = link.get("webUrl") + if result.get("type") != "tv-series" or not web_url: + continue + + url = urljoin("https://www.9now.com.au", web_url) + if url in seen: + continue + seen.add(url) + + yield SearchResult( + id_=url, + title=result.get("name") or result.get("displayName"), + description=result.get("description"), + label="SERIES", + url=url, + ) + + def get_titles(self) -> Movies | Series: + parsed = self._parse_title(self.title) + series_slug = parsed["series"] + + if parsed["kind"] == "episode": + if parsed.get("year"): + episode = self._episode_from_year(series_slug, parsed["year"], parsed["episode"]) + else: + episode = self._episode(series_slug, parsed["season"], parsed["episode"]) + return Series([episode]) + + if parsed["kind"] == "clip": + clip = self._clip(series_slug, parsed["season"], parsed["clip"]) + return Series([clip]) + + episodes = self._series(series_slug) + return Series(episodes) + + def get_tracks(self, title: Movie | Episode) -> Tracks: + brightcove_id = title.id + if not brightcove_id: + raise ValueError("Could not find Brightcove ID for this title") + + data = self._request( + "GET", + self.config["endpoints"]["playback"].format( + account=self.config["brightcove"]["account"], + id=brightcove_id, + ), + headers={ + "BCOV-POLICY": self.config["brightcove"]["policy_key"], + **self.config["headers"], + }, + ) + + title.data["chapters"] = data.get("cue_points") or title.data.get("video", {}).get("cuePoints") + title.data["duration"] = data.get("duration") or title.data.get("video", {}).get("duration") + + source = self._best_source(data.get("sources", [])) + if not source: + raise ValueError("Could not find a playable Brightcove source") + + title.data["license_url"] = ( + (source.get("key_systems") or {}) + .get("com.widevine.alpha", {}) + .get("license_url") + ) + + source_url = source.get("src") + if source.get("type") == "application/dash+xml": + tracks = DASH.from_url(source_url, self.session).to_tracks(title.language) + else: + tracks = HLS.from_url(source_url, self.session).to_tracks(title.language) + + self._add_subtitles(tracks, data.get("text_tracks", [])) + self._mark_descriptive_audio(tracks) + + return tracks + + def get_chapters(self, title: Movie | Episode) -> Chapters: + cue_points = title.data.get("chapters") or [] + duration = title.data.get("duration") + chapters = [] + for cue in cue_points: + timestamp = cue.get("time") + if timestamp and timestamp > 0: + chapters.append(Chapter(timestamp=self._chapter_timestamp_ms(timestamp, duration))) + + return Chapters(chapters) + + def get_widevine_service_certificate(self, **_: Any) -> str: + return None + + def get_widevine_license(self, *, challenge: bytes, title: Movie | Episode, track: Any) -> bytes | str | None: + if license_url := title.data.get("license_url"): + r = self.session.post( + url=license_url, + data=challenge, + headers={ + "Content-Type": "application/octet-stream", + **self.config["headers"], + }, + ) + if r.status_code != 200: + raise ConnectionError(r.text) + return r.content + + return None + + # Service specific + + def _series(self, series_slug: str) -> list[Episode]: + metadata = self._request( + "GET", + self.config["endpoints"]["series"].format(series=series_slug), + params={"device": "web"}, + ) + + seasons = self._extract_seasons(metadata) + if not seasons: + raise ValueError(f"Could not find seasons for title: {series_slug}") + + episodes = [] + for season_slug in seasons: + episodes.extend(self._season_episodes(series_slug, season_slug)) + episodes.extend(self._season_clips(series_slug, season_slug, metadata)) + + return sorted(episodes, key=lambda item: item.data.get("_sort", (0, 0)), reverse=True) + + def _season_episodes(self, series_slug: str, season_slug: str) -> list[Episode]: + data = self._request( + "GET", + self.config["endpoints"]["season_episodes"].format(series=series_slug, season=season_slug), + params={"device": "web"}, + ) + + return [ + self._build_episode(episode, series_slug=series_slug, season_slug=season_slug) + for episode in data.get("episodes", {}).get("items", []) + if (episode.get("video") or {}).get("brightcoveId") + ] + + def _episode(self, series_slug: str, season_slug: str, episode_slug: str) -> Episode: + data = self._request( + "GET", + self.config["endpoints"]["episode"].format( + series=series_slug, + season=season_slug, + episode=episode_slug, + ), + params={"device": "web"}, + ) + episode = data.get("episode") + if not episode: + raise ValueError(f"Could not find episode: {self.title}") + + return self._build_episode(episode, series_slug=series_slug, season_slug=season_slug) + + def _episode_from_year(self, series_slug: str, year: str, episode_slug: str) -> Episode: + episodes = self._series(series_slug) + episode_number = self._slug_number(episode_slug) + episode = next( + ( + item for item in episodes + if item.season == int(year) and item.number == episode_number + ), + None, + ) + if not episode: + episode = next((item for item in episodes if item.number == episode_number), None) + if not episode: + raise ValueError(f"Could not find episode: {self.title}") + + return episode + + def _season_clips(self, series_slug: str, season_slug: str, series_metadata: dict) -> list[Episode]: + data = self._request( + "GET", + self.config["endpoints"]["season"].format(series=series_slug, season=season_slug), + params={"device": "web"}, + ) + clips = self._extract_clips(data) + + if not clips: + clips = [ + clip for clip in self._extract_clips(series_metadata) + if (clip.get("partOfSeason") or {}).get("slug") == season_slug + ] + + clips = sorted(clips, key=self._date_sort_key, reverse=True) + return [ + self._build_clip(clip, series_slug=series_slug, season_slug=season_slug, clip_number=index) + for index, clip in enumerate(clips, start=1) + if (clip.get("video") or {}).get("brightcoveId") + ] + + def _clip(self, series_slug: str, season_slug: str, clip_slug: str) -> Episode: + clips = self._season_clips(series_slug, season_slug, {}) + target = f"/{series_slug}/{season_slug}/{clip_slug}".lower() + clip = next( + ( + item for item in clips + if NINE._same_clip_path((item.data.get("link") or {}).get("webUrl"), target) + ), + None, + ) + if not clip: + raise ValueError(f"Could not find clip: {self.title}") + + return clip + + def _build_episode(self, item: dict, *, series_slug: str, season_slug: str) -> Episode: + season_number = self._season_number(item, season_slug) + episode_number = int(item.get("episodeNumber") or self._slug_number(item.get("slug")) or 0) + + item["_sort"] = (season_number, episode_number) + return Episode( + id_=(item.get("video") or {}).get("brightcoveId"), + service=self.__class__, + title=self._series_name(item, series_slug), + season=season_number, + number=episode_number, + name=item.get("name") or item.get("displayName"), + year=self._year(item), + language="en", + data=item, + ) + + def _build_clip(self, item: dict, *, series_slug: str, season_slug: str, clip_number: int) -> Episode: + season_number = self._season_number(item, season_slug) + item["_sort"] = (season_number, 100000 - clip_number) + + return Episode( + id_=(item.get("video") or {}).get("brightcoveId"), + service=self.__class__, + title=self._series_name(item, series_slug), + season=season_number, + number=0, + name=item.get("displayName") or item.get("name"), + year=self._year(item), + language="en", + data=item, + ) + + def _request(self, method: str, endpoint: str, **kwargs: Any) -> Any: + url = endpoint if endpoint.startswith("http") else urljoin(self.config["endpoints"]["base_url"], endpoint) + + prep = self.session.prepare_request(Request(method, url, **kwargs)) + response = self.session.send(prep) + if response.status_code != 200: + raise ConnectionError(response.text) + + try: + return json.loads(response.content) + except json.JSONDecodeError: + return response.text + + @staticmethod + def _parse_title(title: str) -> dict[str, str]: + title_re = re.compile( + r"^(?:https?://(?:www\.)?9now\.com\.au/)?" + r"(?P[a-z0-9-]+)" + r"(?:/(?Pseason-[^/]+|special|\d{4})" + r"/(?:(?Pepisode-\d+)|(?Pclip-[^/?#]+))" + r"(?:/[^/?#]+)?)?" + r"/?$", + re.IGNORECASE, + ) + + match = title_re.match(title.strip()) + if not match: + raise ValueError(f"Could not parse 9Now title: {title}") + + groups = match.groupdict() + if groups.get("clip"): + groups["kind"] = "clip" + elif groups.get("episode"): + groups["kind"] = "episode" + if groups["season"].isdigit(): + groups["year"] = groups["season"] + else: + groups["year"] = None + else: + groups["kind"] = "series" + groups["year"] = None + + return groups + + @staticmethod + def _same_clip_path(web_url: str | None, target: str) -> bool: + if not web_url: + return False + + path = web_url.lower().rstrip("/") + return path == target or path.startswith(f"{target}/") + + @staticmethod + def _extract_seasons(metadata: dict) -> list[str]: + seasons = [] + for season in metadata.get("seasons", []): + slug = season.get("slug") + if slug and slug not in seasons: + seasons.append(slug) + + return seasons + + @staticmethod + def _extract_clips(metadata: dict) -> list[dict]: + clips = [] + for block in metadata.get("items", []): + for item in block.get("items", []): + if item.get("type") == "clip": + clips.append(item) + + return clips + + @staticmethod + def _best_source(sources: list[dict]) -> dict | None: + dash_drm = next( + ( + source for source in sources + if source.get("type") == "application/dash+xml" + and (source.get("key_systems") or {}).get("com.widevine.alpha") + and source.get("src") + ), + None, + ) + if dash_drm: + return dash_drm + + dash_clear = next( + ( + source for source in sources + if source.get("type") == "application/dash+xml" and source.get("src") + ), + None, + ) + if dash_clear: + return dash_clear + + return next( + ( + source for source in sources + if source.get("type") == "application/x-mpegURL" + and source.get("src", "").startswith("https://") + ), + None, + ) + + @staticmethod + def _add_subtitles(tracks: Tracks, text_tracks: list[dict]) -> None: + for text_track in text_tracks: + source_url = text_track.get("src") + if not source_url and text_track.get("sources"): + source_url = next( + ( + source.get("src") for source in text_track["sources"] + if source.get("src", "").startswith("https://") + ), + text_track["sources"][0].get("src"), + ) + if not source_url: + continue + + tracks.add( + Subtitle( + id_=hashlib.md5(source_url.encode()).hexdigest()[0:6], + url=source_url, + codec=Subtitle.Codec.from_mime(NINE._subtitle_codec(text_track, source_url)), + language=text_track.get("srclang") or "en", + sdh=text_track.get("kind") == "captions", + ) + ) + + @staticmethod + def _subtitle_codec(text_track: dict, source_url: str) -> str: + mime_type = (text_track.get("mime_type") or "").lower() + if "webvtt" in mime_type or source_url.lower().split("?")[0].endswith(".vtt"): + return "vtt" + if "ttml" in mime_type or "xml" in mime_type: + return "ttml" + if "srt" in mime_type or source_url.lower().split("?")[0].endswith(".srt"): + return "srt" + + return source_url.lower().split("?")[0].rsplit(".", 1)[-1] or "vtt" + + @staticmethod + def _chapter_timestamp_ms(timestamp: int | float, duration: int | float | None) -> int: + if duration and timestamp <= duration / 1000: + return int(round(timestamp * 1000)) + + return int(round(timestamp)) + + @staticmethod + def _mark_descriptive_audio(tracks: Tracks) -> None: + for track in tracks.audio: + role = track.data.get("dash", {}).get("representation") + role = role.find("Role") if role is not None else None + if role is not None and role.get("value") in ["description", "alternative", "alternate"]: + track.descriptive = True + + @staticmethod + def _date_sort_key(item: dict) -> datetime: + date = item.get("availability") or item.get("updatedAt") or item.get("airDate") or "" + for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%dT%H:%M:%SZ"): + try: + return datetime.strptime(date, fmt) + except ValueError: + pass + + return datetime.min + + @staticmethod + def _series_name(item: dict, series_slug: str) -> str: + series = item.get("partOfSeries") or {} + return series.get("name") or series_slug.replace("-", " ").title() + + @staticmethod + def _season_number(item: dict, season_slug: str) -> int: + season = item.get("partOfSeason") or {} + if season.get("seasonNumber"): + return int(season["seasonNumber"]) + + if match := re.search(r"season-(\d{4})(?:\d{4})?", season_slug): + return int(match.group(1)) + + if match := re.search(r"season-(\d+)", season_slug): + return int(match.group(1)) + + if season_slug == "special": + return 0 + + return 0 + + @staticmethod + def _slug_number(slug: str | None) -> int: + if not slug: + return 0 + match = re.search(r"(\d+)", slug) + return int(match.group(1)) if match else 0 + + @staticmethod + def _year(item: dict) -> int | None: + date = item.get("airDate") or item.get("availability") or "" + if match := re.match(r"(\d{4})", date): + return int(match.group(1)) + return None diff --git a/packages/envied/src/envied/services/NINE/config.yaml b/packages/envied/src/envied/services/NINE/config.yaml new file mode 100644 index 0000000..7b42c1d --- /dev/null +++ b/packages/envied/src/envied/services/NINE/config.yaml @@ -0,0 +1,18 @@ +headers: + User-Agent: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/139.0.0.0 Safari/537.36" + Accept: "application/json, text/plain, */*" + Origin: "https://www.9now.com.au" + Referer: "https://www.9now.com.au/" + +brightcove: + account: "4460760524001" + policy_key: "BCpkADawqM1TWX5yhWjKdzhXnHCmGvnaozGSDICiEFNRv0fs12m6WA2hLxMHM8TGAEM6pv7lhJsdNhiQi76p4IcsT_jmXdtEU-wnfXhOBTx-cGR7guCqVwjyFAtQa75PFF-TmWESuiYaNTzg" + +endpoints: + base_url: "https://tv-api.9now.com.au" + search: "/v2/pages/search" + series: "/v2/pages/tv-series/{series}" + season: "/v2/pages/tv-series/{series}/seasons/{season}" + season_episodes: "/v2/pages/tv-series/{series}/seasons/{season}/episodes/" + episode: "/v2/pages/tv-series/{series}/seasons/{season}/episodes/{episode}" + playback: "https://edge.api.brightcove.com/playback/v1/accounts/{account}/videos/{id}"