diff --git a/packages/envied/src/envied/services/CR/__init__.py b/packages/envied/src/envied/services/CR/__init__.py index 9e823b6..14fb75c 100644 --- a/packages/envied/src/envied/services/CR/__init__.py +++ b/packages/envied/src/envied/services/CR/__init__.py @@ -1,3 +1,4 @@ +from hashlib import md5 import re import time import uuid @@ -9,11 +10,12 @@ import jwt from langcodes import Language from envied.core.manifests import DASH +from envied.core.manifests.hls import HLS from envied.core.search_result import SearchResult from envied.core.service import Service from envied.core.session import session from envied.core.titles import Episode, Series -from envied.core.tracks import Chapters, Tracks +from envied.core.tracks import Attachment, Chapters, Tracks from envied.core.tracks.chapter import Chapter from envied.core.tracks.subtitle import Subtitle @@ -23,55 +25,70 @@ class CR(Service): Service code for Crunchyroll streaming service (https://www.crunchyroll.com). \b - Version: 3.0.1 - Author: sp4rk.y - Date: 2026-03-26 + Version: 2.1.0 + Author: sp4rk.y & Dex + Date: 2026-02-28 Authorization: Credentials Robustness: Widevine: L3: 1080p, AAC2.0 - + PlayReady: + SL3000: 1080p, AAC2.0 + SL2000: 1080p, AAC2.0 \b Tips: - - Input should be complete URL or series ID + - Input should be complete URL or series ID or Episode ID https://www.crunchyroll.com/series/GRMG8ZQZR/series-name OR GRMG8ZQZR + https://www.crunchyroll.com/en-US/watch/G2XU02X75/ryomen-sukuna OR G2XU02X75 - Supports multiple audio and subtitle languages - Device ID is cached for consistent authentication across runs \b Notes: - - Emulates Android TV client (v3.58.0) with Playback API v3 - - Uses password-based authentication with refresh token caching - - Refresh tokens are cached for 30 days for cross-session reuse + - Uses password-based authentication with token caching - Manages concurrent stream limits automatically """ - TITLE_RE = r"^(?:https?://(?:www\.)?crunchyroll\.com/(?:series|watch)/)?(?P[A-Z0-9]+)" + TITLE_RE = r"^(?:https?://(?:www\.)?crunchyroll\.com/(?:[a-z]{2}(?:-[a-z]{2})?/)?(?:series|watch)/)?(?P[A-Z0-9]+)" LICENSE_LOCK = Lock() MAX_CONCURRENT_STREAMS = 3 ACTIVE_STREAMS: list[tuple[str, str]] = [] - - @staticmethod - def get_session(): - return session("okhttp4") - + + # Reverse map for audio_locale to match 2 chars languages in a_lang param + LANG_MAP: dict = { + "es-LA": "es-419", + "ar-ME": "ar-SA", + "de-DE": "de", + "en-US": "en", + "es-ES": "es", + "fr-FR": "fr", + "hi-IN": "hi", + "it-IT": "it", + "ja-JP": "ja", + "ko-KR": "kr", + "ru-RU": "ru", + "zh-CN": "zh", + "en-GB": "en" + } + @staticmethod @click.command(name="CR", short_help="https://crunchyroll.com") @click.argument("title", type=str, required=True) + @click.option("-hb", "--high-audio-bitrate", is_flag=True, default=False, help="Fetch audio tracks with highest bitrate (192kbps AAC)") + @click.pass_context def cli(ctx, **kwargs) -> "CR": return CR(ctx, **kwargs) - def __init__(self, ctx, title: str): + def __init__(self, ctx, title: str, high_audio_bitrate: bool): + super().__init__(ctx) self.title = title self.account_id: Optional[str] = None self.access_token: Optional[str] = None - self.refresh_token: Optional[str] = None - self.profile_id: Optional[str] = None self.token_expiration: Optional[int] = None self.anonymous_id = str(uuid.uuid4()) - - super().__init__(ctx) + self.audio_lang = ctx.parent.params.get("a_lang") + self.high_audio_bitrate = high_audio_bitrate device_cache_key = "cr_device_id" cached_device = self.cache.get(device_cache_key) @@ -91,6 +108,10 @@ class CR(Service): self.session.headers.update(self.config.get("headers", {})) self.session.headers["etp-anonymous-id"] = self.anonymous_id + @staticmethod + def get_session(): + return session("okhttp4") + @property def auth_header(self) -> dict: """Return authorization header dict.""" @@ -105,23 +126,21 @@ class CR(Service): if cached and not cached.expired: self.access_token = cached.data["access_token"] self.account_id = cached.data.get("account_id") - self.profile_id = cached.data.get("profile_id") - self.refresh_token = cached.data.get("refresh_token") self.token_expiration = cached.data.get("token_expiration") self.session.headers.update(self.auth_header) + self.log.debug("Loaded authentication from cache") else: + self.log.debug("No valid cached token, authenticating") self.authenticate(credential=self.credential) return current_time = int(time.time()) if current_time >= (self.token_expiration - 60): - if self.refresh_token: - self._refresh_access_token() - else: - self.authenticate(credential=self.credential) + self.log.debug("Authentication token expired or expiring soon, re-authenticating") + self.authenticate(credential=self.credential) def authenticate(self, cookies=None, credential=None) -> None: - """Authenticate using username and password credentials, with refresh token support.""" + """Authenticate using username and password credentials.""" super().authenticate(cookies, credential) cache_key = f"cr_auth_token_{credential.sha1 if credential else 'default'}" @@ -130,37 +149,10 @@ class CR(Service): if cached and not cached.expired: self.access_token = cached.data["access_token"] self.account_id = cached.data.get("account_id") - self.profile_id = cached.data.get("profile_id") - self.refresh_token = cached.data.get("refresh_token") self.token_expiration = cached.data.get("token_expiration") else: - refresh_cache_key = f"cr_refresh_token_{credential.sha1 if credential else 'default'}" - refresh_cached = self.cache.get(refresh_cache_key) - - if refresh_cached and not refresh_cached.expired and refresh_cached.data.get("refresh_token"): - self.refresh_token = refresh_cached.data["refresh_token"] - try: - self._refresh_access_token() - self._cache_auth(cache_key) - self.session.headers.update(self.auth_header) - - if self.ACTIVE_STREAMS: - self.ACTIVE_STREAMS.clear() - - try: - self.clear_all_sessions() - except Exception as e: - self.log.warning(f"Failed to clear previous sessions: {e}") - return - except Exception: - self.log.warning("Refresh token expired or invalid, falling back to password login") - self.refresh_token = None - if not credential: - raise ValueError("No credential provided for authentication") - - # Get anonymous client token first (required before password grant) - self._get_client_token() + raise ValueError("Username and password credential required for authentication") response = self.session.post( url=self.config["endpoints"]["token"], @@ -182,18 +174,36 @@ class CR(Service): ) if response.status_code != 200: + self.log.error(f"Login failed: {response.status_code}") try: error_data = response.json() error_msg = error_data.get("error", "Unknown error") error_code = error_data.get("code", "") - raise ValueError(f"Login failed: {response.status_code} - {error_msg} ({error_code})") - except ValueError: - raise + self.log.error(f"Error: {error_msg} ({error_code})") except Exception: - raise ValueError(f"Login failed: {response.status_code} - {response.text}") + self.log.error(f"Response: {response.text}") + response.raise_for_status() - self._apply_token_response(response.json()) - self._cache_auth(cache_key) + token_data = response.json() + self.access_token = token_data["access_token"] + self.account_id = self.get_account_id() + + try: + decoded_token = jwt.decode(self.access_token, options={"verify_signature": False}) + self.token_expiration = decoded_token.get("exp") + except Exception: + self.token_expiration = int(time.time()) + token_data.get("expires_in", 3600) + + cached.set( + data={ + "access_token": self.access_token, + "account_id": self.account_id, + "token_expiration": self.token_expiration, + }, + expiration=self.token_expiration + if isinstance(self.token_expiration, int) and self.token_expiration > int(time.time()) + else 3600, + ) self.session.headers.update(self.auth_header) @@ -205,136 +215,51 @@ class CR(Service): except Exception as e: self.log.warning(f"Failed to clear previous sessions: {e}") - def _apply_token_response(self, token_data: dict) -> None: - """Extract and store auth fields from a token response.""" - self.access_token = token_data["access_token"] - self.refresh_token = token_data.get("refresh_token", self.refresh_token) - self.account_id = token_data.get("account_id", self.account_id) - self.profile_id = token_data.get("profile_id", self.profile_id) - - try: - decoded_token = jwt.decode(self.access_token, options={"verify_signature": False}) - self.token_expiration = decoded_token.get("exp") - except Exception: - self.token_expiration = int(time.time()) + token_data.get("expires_in", 300) - - def _cache_auth(self, cache_key: str) -> None: - """Cache current auth state (access token + refresh token).""" - ttl = 300 - if isinstance(self.token_expiration, int): - remaining = self.token_expiration - int(time.time()) - if remaining > 0: - ttl = remaining - - cached = self.cache.get(cache_key) - cached.set( - data={ - "access_token": self.access_token, - "account_id": self.account_id, - "profile_id": self.profile_id, - "refresh_token": self.refresh_token, - "token_expiration": self.token_expiration, - }, - expiration=ttl, - ) - - # Cache refresh token separately with a long TTL for cross-session reuse - if self.refresh_token: - refresh_cache_key = cache_key.replace("cr_auth_token_", "cr_refresh_token_") - refresh_cached = self.cache.get(refresh_cache_key) - refresh_cached.set( - data={"refresh_token": self.refresh_token}, - expiration=60 * 60 * 24 * 30, # 30 days - ) - - def _get_client_token(self) -> None: - """Get an anonymous client token. Required before password grant on fresh sessions.""" - response = self.session.post( - url=self.config["endpoints"]["token"], - headers={ - "content-type": "application/x-www-form-urlencoded; charset=UTF-8", - }, - data={ - "grant_type": "client_id", - "scope": "offline_access", - "client_id": self.config["client"]["id"], - "client_secret": self.config["client"]["secret"], - }, - ) - if response.status_code == 200: - token_data = response.json() - self.access_token = token_data["access_token"] - self.session.headers.update(self.auth_header) - else: - self.log.warning(f"Failed to get anonymous client token: {response.status_code}") - - def _refresh_access_token(self) -> None: - """Refresh the access token using the stored refresh token.""" - if not self.refresh_token: - raise ValueError("No refresh token available") - - response = self.session.post( - url=self.config["endpoints"]["token"], - headers={ - "content-type": "application/x-www-form-urlencoded; charset=UTF-8", - }, - data={ - "grant_type": "refresh_token", - "refresh_token": self.refresh_token, - "scope": "offline_access", - "client_id": self.config["client"]["id"], - "client_secret": self.config["client"]["secret"], - "device_type": self.device_type, - "device_id": self.device_id, - "device_name": self.device_name, - }, - ) - - if response.status_code != 200: - self.refresh_token = None - raise ValueError(f"Token refresh failed: {response.status_code}") - - self._apply_token_response(response.json()) - - cache_key = f"cr_auth_token_{self.credential.sha1 if self.credential else 'default'}" - self._cache_auth(cache_key) - self.session.headers.update(self.auth_header) - def get_titles(self) -> Union[Series]: """Fetch series and episode information.""" series_id = self.parse_series_id(self.title) - series_http = self.session.get( + series_response = self.session.get( url=self.config["endpoints"]["series"].format(series_id=series_id), params={"locale": self.config["params"]["locale"]}, - ) - series_response = series_http.json() + ).json() - if "error" in series_response: - raise ValueError(f"Series not found: {series_id} - {series_response.get('error')}") + if "code" in series_response: + # Fallback to try with episode id + episode_response = self.session.get( + url=self.config["endpoints"]["episodes"].format(episode_id=series_id), + params={"locale": self.config["params"]["locale"]}, + ).json() + if "code" in episode_response: + raise ValueError(f"Series or episode not found: {series_id}") + + # Extract series id from episodes response + series_id = episode_response["data"][0]["episode_metadata"]["series_id"] + + series_response = self.session.get( + url=self.config["endpoints"]["series"].format(series_id=series_id), + params={"locale": self.config["params"]["locale"]}, + ).json() + + if "code" in series_response: + raise ValueError(f"Series not found from episode metadata: {series_id}") + series_data = ( series_response.get("data", [{}])[0] if isinstance(series_response.get("data"), list) else series_response ) series_title = series_data.get("title", "Unknown Series") - seasons_http = self.session.get( + seasons_response = self.session.get( url=self.config["endpoints"]["seasons"].format(series_id=series_id), - params={ - "locale": self.config["params"]["locale"], - "preferred_audio_language": self.config.get("params", {}).get("preferred_audio_language", "en-US"), - }, - ) - seasons_response = seasons_http.json() + params={"locale": self.config["params"]["locale"]}, + ).json() seasons_data = seasons_response.get("data", []) if not seasons_data: raise ValueError(f"No seasons found for series: {series_id}") - used_season_numbers: set[int] = set() - season_id_to_number: dict[str, int] = {} - all_episode_data = [] special_episodes = [] @@ -342,22 +267,10 @@ class CR(Service): season_id = season["id"] season_number = season.get("season_number", 0) - effective_season_number = season_number - if isinstance(season_number, int) and season_number > 0: - if season_number in used_season_numbers: - candidate = season_number + 1 - while candidate in used_season_numbers: - candidate += 1 - effective_season_number = candidate - used_season_numbers.add(effective_season_number) - - season_id_to_number[season_id] = effective_season_number - - episodes_http = self.session.get( + episodes_response = self.session.get( url=self.config["endpoints"]["season_episodes"].format(season_id=season_id), params={"locale": self.config["params"]["locale"]}, - ) - episodes_response = episodes_http.json() + ).json() episodes_data = episodes_response.get("data", []) @@ -367,7 +280,7 @@ class CR(Service): if episode_number is None or isinstance(episode_number, float): special_episodes.append(episode_data) - all_episode_data.append((episode_data, effective_season_number)) + all_episode_data.append((episode_data, season_number)) if not all_episode_data: raise ValueError(f"No episodes found for series: {series_id}") @@ -421,22 +334,16 @@ class CR(Service): return Series(episodes) - def set_track_metadata( - self, tracks: Tracks, episode_id: str, is_original: bool, audio_locale: Optional[str] = None - ) -> None: + def set_track_metadata(self, tracks: Tracks, episode_id: str, is_original: bool) -> None: """Set metadata for video and audio tracks.""" for video in tracks.videos: video.needs_repack = True video.data["episode_id"] = episode_id video.is_original_lang = is_original - if audio_locale: - video.data["audio_locale"] = audio_locale for audio in tracks.audio: audio.data["episode_id"] = episode_id audio.is_original_lang = is_original - if audio_locale: - audio.data["audio_locale"] = audio_locale - + def get_tracks(self, title: Episode) -> Tracks: """Fetch video, audio, and subtitle tracks for an episode.""" self.ensure_authenticated() @@ -448,12 +355,7 @@ class CR(Service): self.clear_all_sessions() - endpoints_to_try = ["playback", "playback_download"] - - preferred_audio = self.config.get("params", {}).get("preferred_audio_language", "en-US") - initial_response = self.get_playback_data( - episode_id, track_stream=False, endpoint_key="playback", audio_locale=preferred_audio - ) + initial_response = self.get_playback_data(episode_id, track_stream=False) versions = initial_response.get("versions", []) if not versions: @@ -462,136 +364,169 @@ class CR(Service): tracks = None + #endpoints_playback = ["playback"] + + sdh_list = [] for idx, version in enumerate(versions): audio_locale = version.get("audio_locale") version_guid = version.get("guid") is_original = version.get("original", False) - if not audio_locale: + # Verify mapped audio locale to match CR locales + mapped_locale = self.LANG_MAP.get(audio_locale, audio_locale) + if not self.audio_lang or "all" in self.audio_lang or "best" in self.audio_lang or ("orig" in self.audio_lang and is_original): + pass + elif not audio_locale or (audio_locale not in self.audio_lang and mapped_locale not in self.audio_lang): continue request_episode_id = version_guid if version_guid else episode_id - - for endpoint_key in endpoints_to_try: - try: - if idx == 0 and not version_guid and endpoint_key == "playback": - version_response = initial_response - version_token = version_response.get("token") - else: - if idx == 1 and not versions[0].get("guid") and endpoint_key == "playback": - initial_token = initial_response.get("token") - if initial_token: - self.close_stream(episode_id, initial_token) - - try: - version_response = self.get_playback_data( - request_episode_id, - track_stream=False, - endpoint_key=endpoint_key, - audio_locale=audio_locale, - ) - except ValueError as e: - self.log.warning( - f"Could not get playback info for audio {audio_locale} from {endpoint_key}: {e}" - ) - continue - - version_token = version_response.get("token") - - hard_subs = version_response.get("hardSubs", {}) - - root_url = version_response.get("url") - - if root_url and "/clean/" in root_url: - dash_url = root_url - elif "none" in hard_subs: - dash_url = hard_subs["none"].get("url") - elif "fr-FR" in hard_subs: - dash_url = hard_subs["fr-FR"].get("url") - elif hard_subs: - first_key = list(hard_subs.keys())[0] - dash_url = hard_subs[first_key].get("url") - else: - dash_url = None - - if not dash_url: - if version_token: - self.close_stream(request_episode_id, version_token) - continue + + #for endpoint in endpoints_playback: + try: + if idx == 0 and not version_guid: # and endpoint == "playback": + version_response = initial_response + version_token = version_response.get("token") + else: + if idx == 1 and not versions[0].get("guid"): # and endpoint == "playback": + initial_token = initial_response.get("token") + if initial_token: + self.close_stream(episode_id, initial_token) try: - version_tracks = DASH.from_url( - url=dash_url, - session=self.session, - ).to_tracks(language=audio_locale) - - if tracks is None: - tracks = version_tracks - self.set_track_metadata(tracks, request_episode_id, is_original, audio_locale) - else: - self.set_track_metadata(version_tracks, request_episode_id, is_original, audio_locale) - for video in version_tracks.videos: - if not any(v.id == video.id for v in tracks.videos): - tracks.add(video) - for audio in version_tracks.audio: - existing_audio = next((a for a in tracks.audio if a.language == audio.language), None) - if existing_audio is None or ( - hasattr(audio, "bitrate") - and hasattr(existing_audio, "bitrate") - and audio.bitrate > existing_audio.bitrate - ): - tracks.add(audio) - elif existing_audio is None: - tracks.add(audio) - - except Exception as e: - self.log.warning( - f"Failed to parse DASH manifest for audio {audio_locale} from {endpoint_key}: {e}" - ) - if version_token: - self.close_stream(request_episode_id, version_token) + version_response = self.get_playback_data(request_episode_id, track_stream=False) + except ValueError as e: + self.log.warning(f"Could not get playback info for audio {audio_locale}: {e}") continue - if is_original and endpoint_key == "playback": - captions = version_response.get("captions", {}) - subtitles_data = version_response.get("subtitles", {}) - all_subs = {**captions, **subtitles_data} + version_token = version_response.get("token") - for lang_code, sub_data in all_subs.items(): - if lang_code == "none": - continue - - if isinstance(sub_data, dict) and "url" in sub_data: - try: - lang = Language.get(lang_code) - except (ValueError, LookupError): - lang = Language.get("fr") - - subtitle_format = sub_data.get("format", "vtt").lower() - if subtitle_format == "ass" or subtitle_format == "ssa": - codec = Subtitle.Codec.SubStationAlphav4 - else: - codec = Subtitle.Codec.WebVTT - - tracks.add( - Subtitle( - id_=f"subtitle-{audio_locale}-{lang_code}", - url=sub_data["url"], - codec=codec, - language=lang, - forced=False, - sdh=False, - ), - warn_only=True, - ) + hard_subs = version_response.get("hardSubs", {}) + dash_url = None + root_url = version_response.get("url") + + if root_url and "/clean/" in root_url: + dash_url = root_url + + elif "none" in hard_subs: + dash_url = hard_subs["none"].get("url") + + elif hard_subs: + first_key = list(hard_subs.keys())[0] + dash_url = hard_subs[first_key].get("url") + if not dash_url: + self.log.warning(f"No DASH manifest found for audio {audio_locale}, skipping") if version_token: self.close_stream(request_episode_id, version_token) + continue + + try: + + version_tracks = (HLS if "m3u8" in dash_url else DASH).from_url( + url=dash_url, + session=self.session, + ).to_tracks(language=audio_locale) + + # Select high audio bitrate replacing 0 and 1 in mpd url + if self.high_audio_bitrate: + try: + version_tracks.audio.clear(); + + high_bitrate_tracks = (HLS if "m3u8" in dash_url else DASH).from_url( + url=dash_url.replace("/0/", "/1/"), + session=self.session, + ).to_tracks(language=audio_locale) + + version_tracks.audio = high_bitrate_tracks.audio + except Exception as e: + self.log.warning(f"Failed to fetch high bitrate audio for {audio_locale}: {e}") + + if tracks is None: + tracks = version_tracks + self.set_track_metadata(tracks, request_episode_id, is_original) + else: + self.set_track_metadata(version_tracks, request_episode_id, is_original) + # Add videos + for video in version_tracks.videos: + if not any(v.id == video.id for v in tracks.videos): + tracks.add(video) + + # Add audios non-existing in tracks or with higher bitrate + for audio in version_tracks.audio: + if audio.channels == "1.0": + audio.channels = "2.0" # change audio channels for correct channel + + existing_audio = next((a for a in tracks.audio if a.language == audio.language), None) + if existing_audio is None or (hasattr(audio, 'bitrate') and hasattr(existing_audio, 'bitrate') and audio.bitrate > existing_audio.bitrate): + tracks.add(audio) + elif existing_audio is None: + tracks.add(audio) except Exception as e: - self.log.warning(f"Error processing endpoint {endpoint_key} for version {idx}: {e}") + self.log.warning(f"Failed to parse DASH manifest for audio {audio_locale}: {e}") + if version_token: + self.close_stream(request_episode_id, version_token) continue + #if endpoint == "playback": + # Handling subtitles code + try: + lang_audio = [x for x in version_response.get("versions", []) if x["guid"] == version_guid][0]["audio_locale"] + except (IndexError, KeyError, TypeError): + lang_audio = title.language + + captions = version_response.get("captions", {}) + subtitles_data = version_response.get("subtitles", {}) + + if captions: + for subtitle in captions.values(): + + if subtitle["language"] == "none": + continue + + lang = subtitle["language"] + subtitle_format = subtitle.get("format", "vtt").lower() + codec = Subtitle.Codec.SubStationAlphav4 if subtitle_format in ["ass", "ssa"] else Subtitle.Codec.WebVTT + sdh_list.append(lang) + tracks.add( + Subtitle( + id_=md5(subtitle["url"].encode()).hexdigest()[0:6], + url=subtitle["url"], + codec=codec, + language=Language.get(lang), + sdh=True, + ), + warn_only=True, + ) + + if subtitles_data: + for subtitle in subtitles_data.values(): + if subtitle["language"] == "none": + continue + + lang = subtitle["language"] + subtitle_format = subtitle.get("format", "vtt").lower() + codec = Subtitle.Codec.SubStationAlphav4 if subtitle_format in ["ass", "ssa"] else Subtitle.Codec.WebVTT + tracks.add( + Subtitle( + id_=md5(subtitle["url"].encode()).hexdigest()[0:6], + url=subtitle["url"], + codec=codec, + language=Language.get(lang), + forced=False if (str(lang_audio) == str(title.language)) and (lang not in sdh_list) else True, + sdh=False, + ), + warn_only=True, + ) + + if version_token: + self.close_stream(request_episode_id, version_token) + + # continuar si no se pudo procesar el endpoint + except Exception as e: + self.log.warning(f"Error processing version {idx}: {e}") + continue + if versions and versions[0].get("guid"): initial_token = initial_response.get("token") if initial_token: @@ -600,18 +535,74 @@ class CR(Service): if tracks is None: raise ValueError(f"Failed to fetch any tracks for episode: {episode_id}") - for track in tracks.audio + tracks.subtitles: - if track.language: - try: - lang_obj = Language.get(str(track.language)) - base_lang = Language.get(lang_obj.language) - lang_display = base_lang.language_name() - track.name = lang_display - except (ValueError, LookupError): - pass + + images = title.data.get("images", {}) + thumbnails = images.get("thumbnail", []) + if thumbnails: + thumb_variants = thumbnails[0] if isinstance(thumbnails[0], list) else [thumbnails[0]] + if thumb_variants: + thumb_index = min(7, len(thumb_variants) - 1) + thumb = thumb_variants[thumb_index] + if isinstance(thumb, dict) and "source" in thumb: + raw_name = f"{title.name or title.title} - S{title.season:02d}E{title.number:02d}" + thumbnail_name = self.sanitize_filename(raw_name) + tracks.add(Attachment.from_url(url=thumb["source"], name=thumbnail_name)) return tracks + def get_playready_license(self, challenge: bytes, title: Episode, track) -> bytes: + """ + Get PlayReady license for decryption. + + Creates a fresh playback session for each track, gets the license, then immediately + closes the stream. This prevents hitting the 3 concurrent stream limit. + CDN authorization is embedded in the manifest URLs, not tied to active sessions. + """ + self.ensure_authenticated() + + track_episode_id = track.data.get("episode_id", title.id) + + with self.LICENSE_LOCK: + playback_token = None + try: + playback_data = self.get_playback_data(track_episode_id, track_stream=True) + playback_token = playback_data.get("token") + + if not playback_token: + raise ValueError(f"No playback token in response for {track_episode_id}") + + track.data["playback_token"] = playback_token + + license_response = self.session.post( + url=self.config["endpoints"]["license_playready"], + params={"specConform": "true"}, + data=challenge, + headers={ + **self.auth_header, + "content-type": "application/octet-stream", + "accept": "application/octet-stream", + "x-cr-content-id": track_episode_id, + "x-cr-video-token": playback_token, + }, + ) + + if license_response.status_code != 200: + self.log.error(f"License request failed with status {license_response.status_code}") + self.log.error(f"Response: {license_response.text[:500]}") + self.close_stream(track_episode_id, playback_token) + raise ValueError(f"License request failed: {license_response.status_code}") + + self.close_stream(track_episode_id, playback_token) + return license_response.content + + except Exception: + if playback_token: + try: + self.close_stream(track_episode_id, playback_token) + except Exception: + pass + raise + def get_widevine_license(self, challenge: bytes, title: Episode, track) -> bytes: """ Get Widevine license for decryption. @@ -627,8 +618,7 @@ class CR(Service): with self.LICENSE_LOCK: playback_token = None try: - audio_locale = track.data.get("audio_locale") - playback_data = self.get_playback_data(track_episode_id, track_stream=True, audio_locale=audio_locale) + playback_data = self.get_playback_data(track_episode_id, track_stream=True) playback_token = playback_data.get("token") if not playback_token: @@ -648,13 +638,12 @@ class CR(Service): "x-cr-video-token": playback_token, }, ) + if license_response.status_code != 200: + self.log.error(f"License request failed with status {license_response.status_code}") + self.log.error(f"Response: {license_response.text[:500]}") self.close_stream(track_episode_id, playback_token) - try: - error_detail = license_response.text[:200] - except Exception: - error_detail = "Unknown error" - raise ValueError(f"License request failed: {license_response.status_code} - {error_detail}") + raise ValueError(f"License request failed: {license_response.status_code}") self.close_stream(track_episode_id, playback_token) return license_response.content @@ -700,7 +689,6 @@ class CR(Service): chapter_response = self.session.get( url=self.config["endpoints"]["skip_events"].format(episode_id=title.id), ) - special_chapters = [] if chapter_response.status_code == 200: try: @@ -712,54 +700,14 @@ class CR(Service): for chapter_type in ["intro", "recap", "credits", "preview"]: if chapter_info := chapter_data.get(chapter_type): try: - start_time = int(chapter_info["start"] * 1000) - end_time = int(chapter_info.get("end", chapter_info["start"]) * 1000) - special_chapters.append( - { - "start": start_time, - "end": end_time, - "name": chapter_info["type"].capitalize(), - "type": chapter_type, - } + chapters.add( + Chapter( + timestamp=int(chapter_info["start"] * 1000), + name=chapter_info["type"].capitalize(), + ) ) - except Exception: - pass - - special_chapters.sort(key=lambda x: x["start"]) - - all_chapters = [] - chapter_counter = 1 - - all_chapters.append({"timestamp": 0, "name": f"Chapter {chapter_counter}"}) - chapter_counter += 1 - - for idx, special in enumerate(special_chapters): - all_chapters.append({"timestamp": special["start"], "name": special["name"]}) - - should_add_chapter_after = False - - if special["end"] > special["start"]: - if idx + 1 < len(special_chapters): - next_special = special_chapters[idx + 1] - if next_special["start"] - special["end"] > 2000: - should_add_chapter_after = True - else: - should_add_chapter_after = True - - if should_add_chapter_after: - all_chapters.append({"timestamp": special["end"], "name": f"Chapter {chapter_counter}"}) - chapter_counter += 1 - - for chapter in all_chapters: - try: - chapters.add( - Chapter( - timestamp=chapter["timestamp"], - name=chapter["name"], - ) - ) - except Exception: - pass + except Exception as e: + self.log.debug(f"Failed to add {chapter_type} chapter: {e}") return chapters @@ -778,7 +726,8 @@ class CR(Service): ) if response.status_code != 200: - raise ValueError(f"Search request failed with status {response.status_code}") + self.log.error(f"Search request failed with status {response.status_code}") + return search_data = response.json() for result_group in search_data.get("data", []): @@ -806,44 +755,40 @@ class CR(Service): ) except Exception as e: - raise ValueError(f"Search failed: {e}") + self.log.error(f"Search failed: {e}") + return + + def get_account_id(self) -> str: + """Fetch and return the account ID.""" + response = self.session.get(url=self.config["endpoints"]["account_me"], headers=self.auth_header) + + if response.status_code != 200: + self.log.error(f"Failed to get account info: {response.status_code}") + self.log.error(f"Response: {response.text}") + response.raise_for_status() + + data = response.json() + return data["account_id"] def close_stream(self, episode_id: str, token: str) -> None: """Close an active playback stream to free up concurrent stream slots.""" - delete_url = self.config["endpoints"]["playback_delete"].format(episode_id=episode_id, token=token) - closed = False - - for attempt in range(3): - try: - response = self.session.delete(url=delete_url, headers=self.auth_header) - if response.status_code in (200, 204): - closed = True - break - elif response.status_code == 403: - # Auth issue — re-authenticate and retry - if attempt < 2: - self.ensure_authenticated() - continue - self.log.warning(f"Could not close stream for {episode_id}: 403 after re-auth") - break - else: - self.log.warning( - f"Failed to close stream for {episode_id} " - f"(status {response.status_code}, attempt {attempt + 1})" - ) - if attempt < 2: - time.sleep(1) - except Exception as e: - self.log.warning(f"Error closing stream for {episode_id} (attempt {attempt + 1}): {e}") - if attempt < 2: - time.sleep(1) - - # Always remove from local tracking to avoid stale entries - if (episode_id, token) in self.ACTIVE_STREAMS: - self.ACTIVE_STREAMS.remove((episode_id, token)) - - if not closed: - self.log.warning(f"Stream {episode_id}/{token[:12]}... may not have been released server-side") + should_remove = False + try: + response = self.session.delete( + url=self.config["endpoints"]["playback_delete"].format(episode_id=episode_id, token=token), + headers=self.auth_header, + ) + if response.status_code in (200, 204, 403): + should_remove = True + else: + self.log.error( + f"Failed to close stream for {episode_id} (status {response.status_code}): {response.text[:200]}" + ) + except Exception as e: + self.log.error(f"Error closing stream for {episode_id}: {e}") + finally: + if should_remove and (episode_id, token) in self.ACTIVE_STREAMS: + self.ACTIVE_STREAMS.remove((episode_id, token)) def get_active_sessions(self) -> list: """Get all active streaming sessions for the account.""" @@ -854,8 +799,7 @@ class CR(Service): ) if response.status_code == 200: data = response.json() - items = data.get("items", []) - return items + return data.get("items", []) else: self.log.warning(f"Failed to get active sessions (status {response.status_code})") return [] @@ -921,21 +865,13 @@ class CR(Service): return cleared - def get_playback_data( - self, - episode_id: str, - track_stream: bool = True, - endpoint_key: str = "playback", - audio_locale: Optional[str] = None, - ) -> dict: + def get_playback_data(self, episode_id: str, track_stream: bool = True) -> dict: """ Get playback data for an episode with automatic retry on stream limits. Args: episode_id: The episode ID to get playback data for track_stream: Whether to track this stream in active_streams (False for temporary streams) - endpoint_key: Which endpoint to use ('playback' or 'playback_download') - audio_locale: Preferred audio locale (e.g. 'en-US', 'ja-JP') Returns: dict: The playback response data @@ -945,46 +881,44 @@ class CR(Service): """ self.ensure_authenticated() - params: dict[str, str] = {"queue": "false"} - if audio_locale: - params["audio"] = audio_locale - - max_retries = 3 + max_retries = 2 for attempt in range(max_retries + 1): - try: - http_response = self.session.get( - url=self.config["endpoints"][endpoint_key].format(episode_id=episode_id), - params=params, - ) - except Exception as e: - # Session layer exhausted its own retries (e.g. repeated 429s) + response = self.session.get( + url=self.config["endpoints"]["playback"].format(episode_id=episode_id), + params={"queue": "0"}, + ) + + # Handle 429 response status code before json parsing + if response.status_code == 429: + retry_after = int(response.headers.get("Retry-After", 60)) + self.log.warning(f"Rate limited (429). Waiting {retry_after}s before retry... (attempt {attempt + 1}/{max_retries + 1})") if attempt < max_retries: - self.log.warning(f"Playback request failed for {episode_id}: {e}") - self._recover_from_rate_limit(attempt) + cleared = self.clear_all_sessions() + time.sleep(retry_after) continue - raise ValueError(f"Playback request failed for {episode_id} after retries: {e}") + raise ValueError(f"Rate limit exceeded for episode: {episode_id}") - if http_response.status_code == 429: - if attempt < max_retries: - self.log.warning(f"Rate limited (429) on playback for {episode_id}") - self._recover_from_rate_limit(attempt) - continue - raise ValueError(f"Rate limited on playback for {episode_id} after {max_retries} retries") - - try: - response = http_response.json() - except Exception as e: - raise ValueError(f"Playback: failed to parse JSON (episode_id={episode_id}): {e}") - - if "error" in response: + response = response.json() + + if "code" in response: error_code = response.get("code", "") error_msg = response.get("message", response.get("error", "Unknown error")) if error_code == "TOO_MANY_ACTIVE_STREAMS" and attempt < max_retries: self.log.warning(f"Hit stream limit: {error_msg}") - self._recover_from_rate_limit(attempt) + cleared = self.clear_all_sessions() + + if cleared == 0 and attempt == 0: + wait_time = 30 + self.log.warning( + f"Found orphaned sessions from previous run. Waiting {wait_time}s for them to expire..." + ) + time.sleep(wait_time) + continue + self.log.error(f"Playback API error: {error_msg}") + self.log.debug(f"Full response: {response}") raise ValueError(f"Could not get playback info for episode: {episode_id} - {error_msg}") playback_token = response.get("token") @@ -995,19 +929,13 @@ class CR(Service): raise ValueError(f"Failed to get playback data for episode: {episode_id}") - def _recover_from_rate_limit(self, attempt: int) -> None: - """Clear all sessions and wait with exponential backoff before retrying.""" - cleared = self.clear_all_sessions() - wait_time = min(5 * (2**attempt), 30) - if cleared == 0: - wait_time = max(wait_time, 15) - self.log.warning(f"Cleared {cleared} sessions, waiting {wait_time}s before retry...") - time.sleep(wait_time) - def parse_series_id(self, title_input: str) -> str: """Parse series ID from URL or direct ID input.""" match = re.match(self.TITLE_RE, title_input, re.IGNORECASE) if not match: raise ValueError(f"Could not parse series ID from: {title_input}") - series_id = match.group("id") - return series_id + return match.group("id") + + @staticmethod + def sanitize_filename(name): + return re.sub(r'[<>:"/\\|?*¿¡]', '', name).strip() \ No newline at end of file diff --git a/packages/envied/src/envied/services/CR/config.yaml b/packages/envied/src/envied/services/CR/config.yaml index 110483a..a243c6e 100644 --- a/packages/envied/src/envied/services/CR/config.yaml +++ b/packages/envied/src/envied/services/CR/config.yaml @@ -1,7 +1,7 @@ # Crunchyroll API Configuration client: - id: "bludpho6aoup2q0lmw99" - secret: "bEmjmOi1INunmi96t98OFAjtMJq0TwER" + id: "y2arvjb0h0rgvtizlovy" + secret: "JVLvwdIpXvxU-qIBvT1M8oQTr1qlQJX2" # API Endpoints endpoints: @@ -15,20 +15,24 @@ endpoints: # Content Metadata series: "https://www.crunchyroll.com/content/v2/cms/series/{series_id}" seasons: "https://www.crunchyroll.com/content/v2/cms/series/{series_id}/seasons" + episodes: "https://www.crunchyroll.com/content/v2/cms/objects/{episode_id}" season_episodes: "https://www.crunchyroll.com/content/v2/cms/seasons/{season_id}/episodes" skip_events: "https://static.crunchyroll.com/skip-events/production/{episode_id}.json" # Playback - playback: "https://www.crunchyroll.com/playback/v3/{episode_id}/tv/android_tv/play" - playback_download: "https://www.crunchyroll.com/playback/v3/{episode_id}/android/phone/download" - playback_delete: "https://www.crunchyroll.com/playback/v1/token/{episode_id}/{token}" - playback_sessions: "https://www.crunchyroll.com/playback/v1/sessions/streaming" + playback: "https://cr-play-service.prd.crunchyrollsvc.com/v3/{episode_id}/tv/android_tv/play" + #playback_download: "https://cr-play-service.prd.crunchyrollsvc.com/v3/{episode_id}/android/phone/download" + playback_delete: "https://cr-play-service.prd.crunchyrollsvc.com/v1/token/{episode_id}/{token}" + playback_sessions: "https://cr-play-service.prd.crunchyrollsvc.com/v1/sessions/streaming" license_widevine: "https://cr-license-proxy.prd.crunchyrollsvc.com/v1/license/widevine" + license_playready: "https://cr-license-proxy.prd.crunchyrollsvc.com/v1/license/playReady" + + # Discovery search: "https://www.crunchyroll.com/content/v2/discover/search" # Headers for Android TV client headers: - user-agent: "Crunchyroll/ANDROIDTV/3.58.0_22336 (Android 11; en-US; SHIELD Android TV)" + user-agent: "Crunchyroll/ANDROIDTV/3.59.0_22338 (Android 12; en-US; SHIELD Android TV Build/SR1A.211012.001)" accept: "application/json" accept-charset: "UTF-8" accept-encoding: "gzip" @@ -37,8 +41,7 @@ headers: # Query parameters params: - locale: "en-US" - preferred_audio_language: "en-US" + locale: "es-419" # Device parameters for authentication device: diff --git a/packages/envied/src/envied/services/HMAX/__init__.py b/packages/envied/src/envied/services/HMAX/__init__.py old mode 100755 new mode 100644 diff --git a/packages/envied/src/envied/services/HMAX/config.yaml b/packages/envied/src/envied/services/HMAX/config.yaml old mode 100755 new mode 100644 diff --git a/packages/envied/src/envied/services/HMAX/webvtt.py b/packages/envied/src/envied/services/HMAX/webvtt.py new file mode 100644 index 0000000..4bba366 --- /dev/null +++ b/packages/envied/src/envied/services/HMAX/webvtt.py @@ -0,0 +1,341 @@ +import re +import sys +import typing +from typing import Optional + +import pysubs2 +from pycaption import Caption, CaptionList, CaptionNode, CaptionReadError, WebVTTReader, WebVTTWriter + +from envied.core.config import config + + +class CaptionListExt(CaptionList): + @typing.no_type_check + def __init__(self, iterable=None, layout_info=None): + self.first_segment_mpegts = 0 + super().__init__(iterable, layout_info) + + +class CaptionExt(Caption): + @typing.no_type_check + def __init__(self, start, end, nodes, style=None, layout_info=None, segment_index=0, mpegts=0, cue_time=0.0): + style = style or {} + self.segment_index: int = segment_index + self.mpegts: float = mpegts + self.cue_time: float = cue_time + super().__init__(start, end, nodes, style, layout_info) + + +class WebVTTReaderExt(WebVTTReader): + # HLS extension support + RE_TIMESTAMP_MAP = re.compile(r"X-TIMESTAMP-MAP.*") + RE_MPEGTS = re.compile(r"MPEGTS:(\d+)") + RE_LOCAL = re.compile(r"LOCAL:((?:(\d{1,}):)?(\d{2}):(\d{2})\.(\d{3}))") + + def _parse(self, lines: list[str]) -> CaptionList: + captions = CaptionListExt() + start = None + end = None + nodes: list[CaptionNode] = [] + layout_info = None + found_timing = False + segment_index = -1 + mpegts = 0 + cue_time = 0.0 + + # The first segment MPEGTS is needed to calculate the rest. It is possible that + # the first segment contains no cue and is ignored by pycaption, this acts as a fallback. + captions.first_segment_mpegts = 0 + + for i, line in enumerate(lines): + if "-->" in line: + found_timing = True + timing_line = i + last_start_time = captions[-1].start if captions else 0 + try: + start, end, layout_info = self._parse_timing_line(line, last_start_time) + except CaptionReadError as e: + new_msg = f"{e.args[0]} (line {timing_line})" + tb = sys.exc_info()[2] + raise type(e)(new_msg).with_traceback(tb) from None + + elif "" == line: + if found_timing and nodes: + found_timing = False + caption = CaptionExt( + start, + end, + nodes, + layout_info=layout_info, + segment_index=segment_index, + mpegts=mpegts, + cue_time=cue_time, + ) + captions.append(caption) + nodes = [] + + elif "WEBVTT" in line: + # Merged segmented VTT doesn't have index information, track manually. + segment_index += 1 + mpegts = 0 + cue_time = 0.0 + elif m := self.RE_TIMESTAMP_MAP.match(line): + if r := self.RE_MPEGTS.search(m.group()): + mpegts = int(r.group(1)) + + cue_time = self._parse_local(m.group()) + + # Early assignment in case the first segment contains no cue. + if segment_index == 0: + captions.first_segment_mpegts = mpegts + + else: + if found_timing: + if nodes: + nodes.append(CaptionNode.create_break()) + nodes.append(CaptionNode.create_text(self._decode(line))) + else: + # it's a comment or some metadata; ignore it + pass + + # Add a last caption if there are remaining nodes + if nodes: + caption = CaptionExt(start, end, nodes, layout_info=layout_info, segment_index=segment_index, mpegts=mpegts) + captions.append(caption) + + return captions + + @staticmethod + def _parse_local(string: str) -> float: + """ + Parse WebVTT LOCAL time and convert it to seconds. + """ + m = WebVTTReaderExt.RE_LOCAL.search(string) + if not m: + return 0 + + parsed = m.groups() + if not parsed: + return 0 + hours = int(parsed[1]) + minutes = int(parsed[2]) + seconds = int(parsed[3]) + milliseconds = int(parsed[4]) + return (milliseconds / 1000) + seconds + (minutes * 60) + (hours * 3600) + + +def _merge_webvtt_text(vtt_raw: str) -> str: + """ + Pure text-based merge of segmented WebVTT segments. + Preserves cue settings (line:, position:, align:), formatting tags (, ), + and STYLE blocks. Used for DASH segments with absolute timestamps. + """ + segments = re.split(r"(?=WEBVTT)", vtt_raw.strip()) + segments = [s.strip() for s in segments if s.strip()] + + TIMING_RE = re.compile( + r"^(\d{1,2}:\d{2}:\d{2}[.,]\d{3}|\d{2}:\d{2}[.,]\d{3})\s*-->\s*" + r"(\d{1,2}:\d{2}:\d{2}[.,]\d{3}|\d{2}:\d{2}[.,]\d{3})(.*)" + ) + + style_block = None + all_cues: list = [] + + for seg in segments: + seg_lines = seg.splitlines() + i = 0 + if seg_lines and seg_lines[i].startswith("WEBVTT"): + i += 1 + while i < len(seg_lines) and not seg_lines[i].strip(): + i += 1 + # Skip X-TIMESTAMP-MAP line if present + if i < len(seg_lines) and "X-TIMESTAMP-MAP" in seg_lines[i]: + i += 1 + while i < len(seg_lines) and not seg_lines[i].strip(): + i += 1 + # Collect STYLE block (keep only first occurrence) + if i < len(seg_lines) and seg_lines[i].strip() == "STYLE": + style_lines = ["STYLE"] + i += 1 + while i < len(seg_lines) and seg_lines[i].strip(): + style_lines.append(seg_lines[i]) + i += 1 + if style_block is None: + style_block = "\n".join(style_lines) + while i < len(seg_lines) and not seg_lines[i].strip(): + i += 1 + # Parse cues + while i < len(seg_lines): + line = seg_lines[i].strip() + if not line: + i += 1 + continue + if line in ("NOTE", "REGION"): + i += 1 + while i < len(seg_lines) and seg_lines[i].strip(): + i += 1 + continue + m = TIMING_RE.match(line) + if not m: + i += 1 # cue identifier or unknown — skip + continue + start, end, settings = m.group(1), m.group(2), m.group(3) + i += 1 + content_lines: list = [] + while i < len(seg_lines) and seg_lines[i].strip() and "-->" not in seg_lines[i]: + content_lines.append(seg_lines[i].rstrip()) + i += 1 + if content_lines: + text = "\n".join(content_lines) + # Filter out ghost cues: those whose visible text is empty after + # stripping ASS/SSA position tags (e.g. {\an8}) and whitespace. + # Also filter cues with duration < 100ms — these are sync markers + # inserted by some providers (HBO Max, Sky) with no real content. + def _ts_to_ms(ts: str) -> int: + ts = ts.replace(",", ".") + parts = ts.split(":") + if len(parts) == 3: + h, m, s = parts + else: + h, m, s = "0", parts[0], parts[1] + sec, ms = s.split(".") + return (int(h) * 3600 + int(m) * 60 + int(sec)) * 1000 + int(ms[:3].ljust(3, "0")) + visible = re.sub(r"\{[^}]*\}", "", text).strip() + duration_ms = _ts_to_ms(end) - _ts_to_ms(start) + if visible and duration_ms >= 100: + all_cues.append((start, end, settings, text)) + + # Deduplicate exact matches + seen: set = set() + deduped = [] + for cue in all_cues: + key = (cue[0], cue[1], cue[3]) + if key not in seen: + seen.add(key) + deduped.append(cue) + + out = ["WEBVTT", ""] + if style_block: + out.append(style_block) + out.append("") + for s, e, settings, text in deduped: + out.append(f"{s} --> {e}{settings}") + out.append(text) + out.append("") + return "\n".join(out) + + +def merge_segmented_webvtt(vtt_raw: str, segment_durations: Optional[list[int]] = None, timescale: int = 1) -> str: + """ + Merge Segmented WebVTT data. + + Parameters: + vtt_raw: The concatenated WebVTT files to merge. All WebVTT headers must be + appropriately spaced apart, or it may produce unwanted effects like + considering headers as captions, timestamp lines, etc. + segment_durations: A list of each segment's duration. If not provided it will try + to get it from the X-TIMESTAMP-MAP headers, specifically the MPEGTS number. + timescale: The number of time units per second. + + This parses the X-TIMESTAMP-MAP data to compute new absolute timestamps, replacing + the old start and end timestamp values. All X-TIMESTAMP-MAP header information will + be removed from the output as they are no longer of concern. Consider this function + the opposite of a WebVTT Segmenter, a WebVTT Joiner of sorts. + + Algorithm borrowed from N_m3u8DL-RE and shaka-player. + """ + MPEG_TIMESCALE = 90_000 + + # Use pure text merge when timestamps are already absolute — this preserves + # cue settings (line:, position:, align:) and formatting tags (, ). + # Conditions: + # 1. No X-TIMESTAMP-MAP at all (DASH with absolute timestamps) + # 2. X-TIMESTAMP-MAP present but MPEGTS=0 and LOCAL=00:00:00.000 (SKY, some HLS) + has_timestamp_map = 'X-TIMESTAMP-MAP' in vtt_raw + if not has_timestamp_map: + return _merge_webvtt_text(vtt_raw) + # Check if all MPEGTS values are 0 and LOCAL is 00:00:00.000 (already absolute) + import re as _re + all_mpegts = _re.findall(r'MPEGTS:(\d+)', vtt_raw) + all_local = _re.findall(r'LOCAL:([^\s,\n]+)', vtt_raw) + if all_mpegts and all(m == '0' for m in all_mpegts) and all(l == '00:00:00.000' for l in all_local): + return _merge_webvtt_text(vtt_raw) + + # Check config for conversion method preference + conversion_method = config.subtitle.get("conversion_method", "auto") + use_pysubs2 = conversion_method in ("pysubs2", "auto") + + if use_pysubs2: + # Try using pysubs2 first for more lenient parsing + try: + # Use pysubs2 to parse and normalize the VTT + subs = pysubs2.SSAFile.from_string(vtt_raw) + # Convert back to WebVTT string for pycaption processing + normalized_vtt = subs.to_string("vtt") + vtt = WebVTTReaderExt().read(normalized_vtt) + except Exception: + # Fall back to direct pycaption parsing + vtt = WebVTTReaderExt().read(vtt_raw) + else: + # Use pycaption directly + vtt = WebVTTReaderExt().read(vtt_raw) + for lang in vtt.get_languages(): + prev_caption = None + duplicate_index: list[int] = [] + captions = vtt.get_captions(lang) + + # Some providers can produce "segment_index" values that are + # outside the provided segment_durations list after normalization/merge. + # This used to crash with IndexError and abort the entire download. + if segment_durations and captions: + max_idx = max(getattr(c, "segment_index", 0) for c in captions) + if max_idx >= len(segment_durations): + # Pad with the last known duration (or 0 if empty) so indexing is safe. + pad_val = segment_durations[-1] if segment_durations else 0 + segment_durations = segment_durations + [pad_val] * (max_idx - len(segment_durations) + 1) + + if captions[0].segment_index == 0: + first_segment_mpegts = captions[0].mpegts + else: + first_segment_mpegts = segment_durations[0] if segment_durations else captions.first_segment_mpegts + + caption: CaptionExt + for i, caption in enumerate(captions): + # DASH WebVTT doesn't have MPEGTS timestamp like HLS. Instead, + # calculate the timestamp from SegmentTemplate/SegmentList duration. + likely_dash = first_segment_mpegts == 0 and caption.mpegts == 0 + if likely_dash and segment_durations: + # Defensive: segment_index can still be out of range if captions are malformed. + if caption.segment_index < 0 or caption.segment_index >= len(segment_durations): + continue + duration = segment_durations[caption.segment_index] + caption.mpegts = MPEG_TIMESCALE * (duration / timescale) + + if caption.mpegts == 0: + continue + + # Commeted to fix DSNP subs being out of sync and mistimed. + # seconds = (caption.mpegts - first_segment_mpegts) / MPEG_TIMESCALE - caption.cue_time + # offset = seconds * 1_000_000 # pycaption use microseconds + + # if caption.start < offset: + # caption.start += offset + # caption.end += offset + + # If the difference between current and previous captions is <=1ms + # and the payload is equal then splice. + if ( + prev_caption + and not caption.is_empty() + and (caption.start - prev_caption.end) <= 1000 # 1ms in microseconds + and caption.get_text() == prev_caption.get_text() + ): + prev_caption.end = caption.end + duplicate_index.append(i) + + prev_caption = caption + + # Remove duplicate + captions[:] = [c for c_index, c in enumerate(captions) if c_index not in set(duplicate_index)] + + return WebVTTWriter().write(vtt)