diff --git a/packages/envied/src/envied/services/SBS/__init__.py b/packages/envied/src/envied/services/SBS/__init__.py new file mode 100644 index 0000000..d0c526b --- /dev/null +++ b/packages/envied/src/envied/services/SBS/__init__.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Generator +from urllib.parse import urljoin, urlparse +from typing import Any + +import click +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, Movie, Movies, Series +from envied.core.tracks import Chapters, Subtitle, Tracks +from envied.core.utils.xml import load_xml + + +class SBS(Service): + """ + \b + Service code for SBS ondemand streaming service (https://www.sbs.com.au/ondemand/). + + \b + Version: 1.0.1 + Author: stabbedbybrick + Authorization: None + Geofence: AU (API and downloads) + Robustness: + AES: 720p, AAC2.0 + + \b + Tips: + - Input should be complete URL: + SERIES: https://www.sbs.com.au/ondemand/tv-series/reckless + EPISODE: https://www.sbs.com.au/ondemand/tv-series/reckless/season-1/reckless-s1-ep1/2459384899653 + MOVIE: https://www.sbs.com.au/ondemand/movie/silence/1363535939614 + SPORT: https://www.sbs.com.au/ondemand/sports-series/australian-championship-2025/football-australian-championship-2025/australian-championship-2025-s2025-ep40/2457638979614 + + \b + Notes: + - SBS uses transport streams for HLS, meaning the video and audio are a part of the same stream. + As a result only videos are listed as tracks, but the audio will be included as well. + + """ + + GEOFENCE = ("au",) + + @staticmethod + @click.command(name="SBS", short_help="https://www.sbs.com.au/ondemand/", help=__doc__) + @click.argument("title", type=str) + @click.pass_context + def cli(ctx: Context, **kwargs: Any) -> SBS: + return SBS(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]: + params = { + "q": self.title.strip(), + } + + results = self._request("GET", "https://content-search.pr.sbsod.com/catalogue", params=params)["items"] + + for result in results: + if result.get("entityType") in ("PAGE"): + continue + + label = result.get("entityType") + slug = result.get("slug") + title = result.get("title") + description = result.get("description") + yield SearchResult( + id_=f"https://www.sbs.com.au/ondemand/{label}/{slug}", + title=title, + description=description, + label=label, + url=f"https://www.sbs.com.au/ondemand/{label}/{slug}", + ) + + + def get_titles(self) -> Movies | Series: + regex = re.compile( + r"^https://www.sbs.com.au/ondemand/" + r"(?Ptv-series|tv-program|sports-series|movie|watch)" + r"(?:/|/.*/)" + r"(?P[^/]+)/?$" + ) + + match = regex.search(self.title) + if not match: + raise ValueError(f"Invalid URL input: {self.title}") + + entity_type, entity_id = (match.group(i) for i in ("entity", "id")) + + if entity_type in ("movie", "tv-program") and entity_id.isdigit(): + movie = self._movie(entity_id) + return Movies(movie) + + elif entity_id.isdigit(): + episode = self._episode(entity_id) + return Series(episode) + + elif entity_type in ("tv-series", "sports-series"): + episodes = self._series(urlparse(self.title).path) + return Series(episodes) + + def get_tracks(self, title: Movie | Episode) -> Tracks: + smil = self._request("GET", f"/api/v3/video_smil?id={title.id}") + + body = load_xml(smil).find("body").find("seq") + section = body.find("par") or body + + manifest = next((x.get("src") for x in section.findall("video")), None) + subtitles = [(x.get("src"), x.get("lang"), x.get("type")) for x in section.findall("textstream")] + + tracks = HLS.from_url(manifest, self.session).to_tracks(title.language) + + if subtitles: + for url, lang, type in subtitles: + if "ttaf+xml" in type: + continue + codec = type.split("/")[-1] + tracks.add( + Subtitle( + id_=hashlib.md5(url.encode()).hexdigest()[0:6], + url=url, + codec=Subtitle.Codec.from_mime(codec), + language=lang, + sdh="_CC" in url, + ) + ) + + return tracks + + def get_chapters(self, title: Movie | Episode) -> Chapters: + return Chapters() + + # Service specific + + def _series(self, path: str) -> Episode: + if "ondemand" in path: + path = path.split("ondemand")[1] + + metadata = self._request("GET", f"https://catalogue.pr.sbsod.com{path}") + + seasons = metadata.get("seasons") + if not seasons: + raise ValueError(f"Failed to find seasons for title: {path}") + + episodes = [] + for season in seasons: + for episode in season.get("episodes"): + episodes.append( + Episode( + id_=episode.get("mpxMediaID"), + service=self.__class__, + title=episode.get("seriesTitle"), + season=int(episode.get("seasonNumber", 0)), + number=int(episode.get("episodeNumber", 0)), + name=episode.get("title"), + year=episode.get("releaseYear"), + language=metadata.get("localeID") or "en", + data=episode, + ) + ) + + return episodes + + def _movie(self, entity_id: str) -> Movie: + metadata = self._request("GET", f"https://catalogue.pr.sbsod.com/mpx-media/{entity_id}") + + return [ + Movie( + id_=metadata.get("mpxMediaID"), + service=self.__class__, + name=metadata.get("title") or metadata.get("cdpTitle"), + year=metadata.get("releaseYear"), + language=metadata.get("localeID") or "en", + data=metadata, + ) + ] + + def _episode(self, entity_id: str) -> Episode: + metadata = self._request("GET", f"https://catalogue.pr.sbsod.com/mpx-media/{entity_id}") + + return [ + Episode( + id_=metadata.get("mpxMediaID"), + service=self.__class__, + title=metadata.get("seriesTitle"), + season=int(metadata.get("seasonNumber", 0)), + number=int(metadata.get("episodeNumber", 0)), + name=metadata.get("title") or metadata.get("cdpTitle"), + year=metadata.get("releaseYear"), + language=metadata.get("localeID") or "en", + data=metadata, + ) + ] + + def _request(self, method: str, endpoint: str, **kwargs: Any) -> Any[dict | str]: + url = 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(f"{response.text}") + + try: + return json.loads(response.content) + + except json.JSONDecodeError: + return response.text + + except ValueError as e: + raise ValueError(f"Failed to parse JSON: {response.text}") from e + diff --git a/packages/envied/src/envied/services/SBS/config.yaml b/packages/envied/src/envied/services/SBS/config.yaml new file mode 100644 index 0000000..fad5044 --- /dev/null +++ b/packages/envied/src/envied/services/SBS/config.yaml @@ -0,0 +1,7 @@ +headers: + User-Agent: "AndroidTV/!/!" + +endpoints: + base_url: "https://www.sbs.com.au" + dai: "http://pubads.g.doubleclick.net/ondemand/hls/content/2488267/vid/{vid}/streams" + catalogue: "https://catalogue.pr.sbsod.com" diff --git a/packages/envied/src/envied/services/TUBI/__init__.py b/packages/envied/src/envied/services/TUBI/__init__.py index 3a2bfd9..6576c83 100644 --- a/packages/envied/src/envied/services/TUBI/__init__.py +++ b/packages/envied/src/envied/services/TUBI/__init__.py @@ -26,7 +26,7 @@ class TUBI(Service): Service code for TubiTV streaming service (https://tubitv.com/) \b - Version: 1.0.5 + Version: 1.0.6 Author: stabbedbybrick Authorization: Cookies (Optional) Geofence: Locked to whatever region the user is in (API only) @@ -54,7 +54,7 @@ class TUBI(Service): - Search is currently disabled. """ - TITLE_RE = r"^(?:https?://(?:www\.)?tubitv\.com?)?/(?Pmovies|series|tv-shows)/(?P[a-z0-9-]+)" + TITLE_RE = r"^(?:https?://(?:www\.)?tubitv\.com?)?/(?:[a-z]{2}-[a-z]{2}/)?(?Pmovies|series|tv-shows)/(?P[a-z0-9-]+)" @staticmethod @click.command(name="TUBI", short_help="https://tubitv.com/", help=__doc__) @@ -276,6 +276,9 @@ class TUBI(Service): if cue_points.get("postlude"): chapters.append(Chapter(name="End Credits", timestamp=float(cue_points["postlude"]))) + 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) def get_widevine_service_certificate(self, **_: Any) -> str: