service update

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"(?P<entity>tv-series|tv-program|sports-series|movie|watch)"
+            r"(?:/|/.*/)"
+            r"(?P<id>[^/]+)/?$"
+        )
+
+        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
+

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"

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?)?/(?P<type>movies|series|tv-shows)/(?P<id>[a-z0-9-]+)"
+    TITLE_RE = r"^(?:https?://(?:www\.)?tubitv\.com?)?/(?:[a-z]{2}-[a-z]{2}/)?(?P<type>movies|series|tv-shows)/(?P<id>[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: