From 93ef79441253ff23f3cf088b36a0822f8d45304b Mon Sep 17 00:00:00 2001 From: kenzuyaa Date: Tue, 2 Sep 2025 12:30:54 +0700 Subject: [PATCH] refactor(Netflix): extract track hydration logic into separate method - Moved hydration of unavailable audio and subtitle tracks into hydrate_all_tracks method - Replaced large inline hydration code with a single call to hydrate_all_tracks - Improved clarity by encapsulating hydration steps including logging and error handling - Maintained original behavior with detailed debug and warning messages - Added comprehensive parameters to hydrate_all_tracks for audio, subtitle, primary tracks, and language context - Ensured hydration method returns complete Tracks object for easier track management and addition --- unshackle/services/Netflix/__init__.py | 325 +++++++++++++------------ 1 file changed, 175 insertions(+), 150 deletions(-) diff --git a/unshackle/services/Netflix/__init__.py b/unshackle/services/Netflix/__init__.py index d9b5b0c..e7d8f55 100644 --- a/unshackle/services/Netflix/__init__.py +++ b/unshackle/services/Netflix/__init__.py @@ -895,156 +895,14 @@ class Netflix(Service): # Hydrate missing tracks if unavailable_audio_tracks or unavailable_subtitle: - # Show hydration information once - audio_count = len(unavailable_audio_tracks) - subtitle_count = len(unavailable_subtitle) - - hydration_parts = [] - if audio_count > 0: - hydration_parts.append(f"audio ({audio_count})") - if subtitle_count > 0: - hydration_parts.append(f"subtitle ({subtitle_count})") - - hydration_info = " and ".join(hydration_parts) - self.log.info(f"Hydrating {hydration_info} tracks. Total: {audio_count + subtitle_count}") - - # Handle mismatched lengths - use last successful tracks when needed - last_successful_subtitle = self._get_empty_track_tuple() if not unavailable_subtitle else unavailable_subtitle[-1] - last_successful_audio = self._get_empty_track_tuple() if not unavailable_audio_tracks else unavailable_audio_tracks[-1] - - # For subtitle-only hydration, use primary audio track if available - primary_audio_for_subtitle_hydration = primary_audio_tracks[0] if primary_audio_tracks and not unavailable_audio_tracks and unavailable_subtitle else self._get_empty_track_tuple() - - # Process audio tracks first, then handle subtitles separately if needed - max_length = max(len(unavailable_audio_tracks), len(unavailable_subtitle)) - - for hydration_index in range(max_length): - # Get audio track info for this index, or use last successful one if available - if hydration_index < len(unavailable_audio_tracks): - audio_hydration = unavailable_audio_tracks[hydration_index] - is_real_audio_request = True # This is a real audio to be added to tracks - elif unavailable_audio_tracks: # Use last successful audio track for context only - audio_hydration = last_successful_audio - is_real_audio_request = False # This is just for context, don't add to tracks - elif primary_audio_for_subtitle_hydration[0] is not None: # Use primary audio for subtitle hydration - audio_hydration = primary_audio_for_subtitle_hydration - is_real_audio_request = False # This is just for context, don't add to tracks - self.log.debug(f"Using primary audio track for subtitle hydration: {audio_hydration[1]}") - else: - audio_hydration = self._get_empty_track_tuple() - is_real_audio_request = False - - # Get subtitle track info for this index, or use last successful one if available - if hydration_index < len(unavailable_subtitle): - subtitle_hydration = unavailable_subtitle[hydration_index] - is_real_subtitle_request = True # This is a real subtitle to be added to tracks - elif unavailable_subtitle: # Use last successful subtitle track for context only - subtitle_hydration = last_successful_subtitle - is_real_subtitle_request = False # This is just for context, don't add to tracks - else: - subtitle_hydration = self._get_empty_track_tuple() - is_real_subtitle_request = False - - try: - - - # Prepare track IDs for API request - convert None to None for proper API handling - audio_track_id = audio_hydration[0] if audio_hydration[0] is not None else None - subtitle_track_id = subtitle_hydration[0] if subtitle_hydration[0] is not None else None - - # Log what we're trying to hydrate - self._log_hydration_attempt(hydration_index, audio_hydration, subtitle_hydration, - is_real_audio_request, is_real_subtitle_request) - - # Only call get_manifest if we have valid tracks to hydrate - should_hydrate_audio = self._is_valid_track_for_hydration(audio_hydration) - - if not should_hydrate_audio: - self.log.debug(f"Skipping hydration at index {hydration_index} - no audio tracks to hydrate") - continue - - # If we still don't have a subtitle track ID, skip this hydration to avoid API error - if subtitle_track_id is None: - self.log.warning(f"Skipping hydration at index {hydration_index} - no subtitle track available for API request context") - continue - - - - hydrated_manifest = self.get_manifest(title, self.profiles, subtitle_track_id, audio_track_id) - - # Handle hydrated audio tracks (only if it's a real audio request, not reused) - if is_real_audio_request and should_hydrate_audio and "audio_tracks" in hydrated_manifest: - try: - audios = next((item for item in hydrated_manifest["audio_tracks"] if 'id' in item and item["id"] == audio_hydration[1]), None) - if audios and "streams" in audios: - audio_lang = audios.get("language", "unknown") - self.log.debug(f"Processing hydrated audio track_id: {audio_hydration[1]}, language: {audio_lang}, streams_count: {len(audios['streams'])}") - for stream_index, stream in enumerate(audios["streams"]): - try: - stream_id = stream.get("downloadable_id", "unknown") - tracks.add( - Audio( - id_=stream["downloadable_id"], - url=stream["urls"][0]["url"], - codec=Audio.Codec.from_netflix_profile(stream["content_profile"]), - language=Language.get(self.NF_LANG_MAP.get(audios["language"]) or audios["language"]), - is_original_lang=audios["language"] == original_language.language, - bitrate=stream["bitrate"] * 1000, - channels=stream["channels"], - descriptive=audios.get("rawTrackType", "").lower() == "assistive", - name="[Original]" if Language.get(audios["language"]).language == original_language.language else None, - joc=16 if "atmos" in stream["content_profile"] else None - ) - ) - except Exception as e: - stream_id = stream.get("downloadable_id", "unknown") if isinstance(stream, dict) else "unknown" - self.log.warning(f"Failed to process hydrated audio stream at hydration_index {hydration_index}, stream_index {stream_index}, audio_track_id: {audio_hydration[1]}, stream_id: {stream_id}, error: {e}") - continue - else: - self.log.warning(f"No audio streams found for hydrated audio_track_id: {audio_hydration[1]} at hydration_index {hydration_index}") - except Exception as e: - self.log.warning(f"Failed to find hydrated audio track at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1]}, error: {e}") - elif not is_real_audio_request and audio_hydration[1] is not None: - self.log.debug(f"Used audio track context for API request at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1]} (not adding to tracks)") - - # Handle hydrated subtitle tracks (only if it's a real subtitle request, not reused) - if is_real_subtitle_request and self._is_valid_track_for_hydration(subtitle_hydration) and "timedtexttracks" in hydrated_manifest: - try: - subtitles = next((item for item in hydrated_manifest["timedtexttracks"] if 'id' in item and item["id"] == subtitle_hydration[1]), None) - if subtitles and "downloadableIds" in subtitles and "ttDownloadables" in subtitles: - subtitle_lang = subtitles.get("language", "unknown") - self.log.debug(f"Processing hydrated subtitle track_id: {subtitle_hydration[1]}, language: {subtitle_lang}") - - id = list(subtitles["downloadableIds"].values()) - if id: - language = Language.get(subtitles["language"]) - profile = next(iter(subtitles["ttDownloadables"].keys())) - tt_downloadables = next(iter(subtitles["ttDownloadables"].values())) - tracks.add( - Subtitle( - id_=id[0], - url=tt_downloadables["urls"][0]["url"], - codec=Subtitle.Codec.from_netflix_profile(profile), - language=language, - forced=subtitles.get("isForcedNarrative", False), - cc=subtitles.get("rawTrackType") == "closedcaptions", - sdh=subtitles.get("trackVariant") == 'STRIPPED_SDH' if "trackVariant" in subtitles else False, - is_original_lang=subtitles.get("language") == original_language.language, - name=("[Original]" if language.language == original_language.language else None or "[Dubbing]" if "trackVariant" in subtitles and subtitles["trackVariant"] == "DUBTITLE" else None), - ) - ) - else: - self.log.warning(f"No downloadable IDs found for hydrated subtitle_track_id: {subtitle_hydration[1]} at hydration_index {hydration_index}") - else: - self.log.warning(f"No subtitle data found for hydrated subtitle_track_id: {subtitle_hydration[1]} at hydration_index {hydration_index}") - except Exception as e: - self.log.warning(f"Failed to process hydrated subtitle track at hydration_index {hydration_index}, subtitle_track_id: {subtitle_hydration[1]}, error: {e}") - elif not is_real_subtitle_request and subtitle_hydration[1] is not None: - self.log.debug(f"Used subtitle track context for API request at hydration_index {hydration_index}, subtitle_track_id: {subtitle_hydration[1]} (not adding to tracks)") - - except Exception as e: - self.log.warning(f"Failed to hydrate tracks at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1] or 'None'}, subtitle_track_id: {subtitle_hydration[1] or 'None'}, error: {e}") - continue + hydrated_tracks = self.hydrate_all_tracks( + title=title, + unavailable_audio_tracks=unavailable_audio_tracks, + unavailable_subtitle=unavailable_subtitle, + primary_audio_tracks=primary_audio_tracks, + original_language=original_language + ) + tracks.add(hydrated_tracks) else: self.log.info("No tracks need hydration") @@ -1103,6 +961,173 @@ class Netflix(Service): """Return an empty track tuple with None values.""" return (None, None) + def hydrate_all_tracks(self, title: Title_T, unavailable_audio_tracks: List[Tuple[str, str]], + unavailable_subtitle: List[Tuple[str, str]], primary_audio_tracks: List[Tuple[str, str]], + original_language: Language) -> Tracks: + """ + Hydrate all missing audio and subtitle tracks. + + Args: + title: The title object for which to hydrate tracks + unavailable_audio_tracks: List of audio track tuples (new_track_id, id) that need hydration + unavailable_subtitle: List of subtitle track tuples (new_track_id, id) that need hydration + primary_audio_tracks: List of primary audio track tuples for context in subtitle hydration + original_language: The original language of the content + + Returns: + Tracks: A Tracks object containing all hydrated audio and subtitle tracks + """ + hydrated_tracks = Tracks() + + # Show hydration information once + audio_count = len(unavailable_audio_tracks) + subtitle_count = len(unavailable_subtitle) + + hydration_parts = [] + if audio_count > 0: + hydration_parts.append(f"audio ({audio_count})") + if subtitle_count > 0: + hydration_parts.append(f"subtitle ({subtitle_count})") + + hydration_info = " and ".join(hydration_parts) + self.log.info(f"Hydrating {hydration_info} tracks. Total: {audio_count + subtitle_count}") + + # Handle mismatched lengths - use last successful tracks when needed + last_successful_subtitle = self._get_empty_track_tuple() if not unavailable_subtitle else unavailable_subtitle[-1] + last_successful_audio = self._get_empty_track_tuple() if not unavailable_audio_tracks else unavailable_audio_tracks[-1] + + # For subtitle-only hydration, use primary audio track if available + primary_audio_for_subtitle_hydration = primary_audio_tracks[0] if primary_audio_tracks and not unavailable_audio_tracks and unavailable_subtitle else self._get_empty_track_tuple() + + # Process audio tracks first, then handle subtitles separately if needed + max_length = max(len(unavailable_audio_tracks), len(unavailable_subtitle)) + + for hydration_index in range(max_length): + # Get audio track info for this index, or use last successful one if available + if hydration_index < len(unavailable_audio_tracks): + audio_hydration = unavailable_audio_tracks[hydration_index] + is_real_audio_request = True # This is a real audio to be added to tracks + elif unavailable_audio_tracks: # Use last successful audio track for context only + audio_hydration = last_successful_audio + is_real_audio_request = False # This is just for context, don't add to tracks + elif primary_audio_for_subtitle_hydration[0] is not None: # Use primary audio for subtitle hydration + audio_hydration = primary_audio_for_subtitle_hydration + is_real_audio_request = False # This is just for context, don't add to tracks + self.log.debug(f"Using primary audio track for subtitle hydration: {audio_hydration[1]}") + else: + audio_hydration = self._get_empty_track_tuple() + is_real_audio_request = False + + # Get subtitle track info for this index, or use last successful one if available + if hydration_index < len(unavailable_subtitle): + subtitle_hydration = unavailable_subtitle[hydration_index] + is_real_subtitle_request = True # This is a real subtitle to be added to tracks + elif unavailable_subtitle: # Use last successful subtitle track for context only + subtitle_hydration = last_successful_subtitle + is_real_subtitle_request = False # This is just for context, don't add to tracks + else: + subtitle_hydration = self._get_empty_track_tuple() + is_real_subtitle_request = False + + try: + # Prepare track IDs for API request - convert None to None for proper API handling + audio_track_id = audio_hydration[0] if audio_hydration[0] is not None else None + subtitle_track_id = subtitle_hydration[0] if subtitle_hydration[0] is not None else None + + # Log what we're trying to hydrate + self._log_hydration_attempt(hydration_index, audio_hydration, subtitle_hydration, + is_real_audio_request, is_real_subtitle_request) + + # Only call get_manifest if we have valid tracks to hydrate + should_hydrate_audio = self._is_valid_track_for_hydration(audio_hydration) + + if not should_hydrate_audio: + self.log.debug(f"Skipping hydration at index {hydration_index} - no audio tracks to hydrate") + continue + + # If we still don't have a subtitle track ID, skip this hydration to avoid API error + if subtitle_track_id is None: + self.log.warning(f"Skipping hydration at index {hydration_index} - no subtitle track available for API request context") + continue + + hydrated_manifest = self.get_manifest(title, self.profiles, subtitle_track_id, audio_track_id) + + # Handle hydrated audio tracks (only if it's a real audio request, not reused) + if is_real_audio_request and should_hydrate_audio and "audio_tracks" in hydrated_manifest: + try: + audios = next((item for item in hydrated_manifest["audio_tracks"] if 'id' in item and item["id"] == audio_hydration[1]), None) + if audios and "streams" in audios: + audio_lang = audios.get("language", "unknown") + self.log.debug(f"Processing hydrated audio track_id: {audio_hydration[1]}, language: {audio_lang}, streams_count: {len(audios['streams'])}") + for stream_index, stream in enumerate(audios["streams"]): + try: + stream_id = stream.get("downloadable_id", "unknown") + hydrated_tracks.add( + Audio( + id_=stream["downloadable_id"], + url=stream["urls"][0]["url"], + codec=Audio.Codec.from_netflix_profile(stream["content_profile"]), + language=Language.get(self.NF_LANG_MAP.get(audios["language"]) or audios["language"]), + is_original_lang=audios["language"] == original_language.language, + bitrate=stream["bitrate"] * 1000, + channels=stream["channels"], + descriptive=audios.get("rawTrackType", "").lower() == "assistive", + name="[Original]" if Language.get(audios["language"]).language == original_language.language else None, + joc=16 if "atmos" in stream["content_profile"] else None + ) + ) + except Exception as e: + stream_id = stream.get("downloadable_id", "unknown") if isinstance(stream, dict) else "unknown" + self.log.warning(f"Failed to process hydrated audio stream at hydration_index {hydration_index}, stream_index {stream_index}, audio_track_id: {audio_hydration[1]}, stream_id: {stream_id}, error: {e}") + continue + else: + self.log.warning(f"No audio streams found for hydrated audio_track_id: {audio_hydration[1]} at hydration_index {hydration_index}") + except Exception as e: + self.log.warning(f"Failed to find hydrated audio track at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1]}, error: {e}") + elif not is_real_audio_request and audio_hydration[1] is not None: + self.log.debug(f"Used audio track context for API request at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1]} (not adding to tracks)") + + # Handle hydrated subtitle tracks (only if it's a real subtitle request, not reused) + if is_real_subtitle_request and self._is_valid_track_for_hydration(subtitle_hydration) and "timedtexttracks" in hydrated_manifest: + try: + subtitles = next((item for item in hydrated_manifest["timedtexttracks"] if 'id' in item and item["id"] == subtitle_hydration[1]), None) + if subtitles and "downloadableIds" in subtitles and "ttDownloadables" in subtitles: + subtitle_lang = subtitles.get("language", "unknown") + self.log.debug(f"Processing hydrated subtitle track_id: {subtitle_hydration[1]}, language: {subtitle_lang}") + + id = list(subtitles["downloadableIds"].values()) + if id: + language = Language.get(subtitles["language"]) + profile = next(iter(subtitles["ttDownloadables"].keys())) + tt_downloadables = next(iter(subtitles["ttDownloadables"].values())) + hydrated_tracks.add( + Subtitle( + id_=id[0], + url=tt_downloadables["urls"][0]["url"], + codec=Subtitle.Codec.from_netflix_profile(profile), + language=language, + forced=subtitles.get("isForcedNarrative", False), + cc=subtitles.get("rawTrackType") == "closedcaptions", + sdh=subtitles.get("trackVariant") == 'STRIPPED_SDH' if "trackVariant" in subtitles else False, + is_original_lang=subtitles.get("language") == original_language.language, + name=("[Original]" if language.language == original_language.language else None or "[Dubbing]" if "trackVariant" in subtitles and subtitles["trackVariant"] == "DUBTITLE" else None), + ) + ) + else: + self.log.warning(f"No downloadable IDs found for hydrated subtitle_track_id: {subtitle_hydration[1]} at hydration_index {hydration_index}") + else: + self.log.warning(f"No subtitle data found for hydrated subtitle_track_id: {subtitle_hydration[1]} at hydration_index {hydration_index}") + except Exception as e: + self.log.warning(f"Failed to process hydrated subtitle track at hydration_index {hydration_index}, subtitle_track_id: {subtitle_hydration[1]}, error: {e}") + elif not is_real_subtitle_request and subtitle_hydration[1] is not None: + self.log.debug(f"Used subtitle track context for API request at hydration_index {hydration_index}, subtitle_track_id: {subtitle_hydration[1]} (not adding to tracks)") + + except Exception as e: + self.log.warning(f"Failed to hydrate tracks at hydration_index {hydration_index}, audio_track_id: {audio_hydration[1] or 'None'}, subtitle_track_id: {subtitle_hydration[1] or 'None'}, error: {e}") + continue + + return hydrated_tracks + def _log_hydration_attempt(self, hydration_index: int, audio_data: tuple, subtitle_data: tuple, is_real_audio: bool, is_real_subtitle: bool) -> None: """Log hydration attempt details."""