unshackle-dl/unshackle
1
0
Fork 0
forked from kenzuya/unshackle
Code Pull Requests 1 Activity

refactor(netflix): unify DRM handling and improve track hydration logic

Browse Source 
- Added DRM system detection method to distinguish Widevine and PlayReady CDMs
- Implemented create_drm method to instantiate appropriate DRM objects based on system
- Updated track hydration to add all tracks on first profile/range, only videos otherwise
- Changed get_playready_license to reuse Widevine license retrieval method
- Replaced direct Widevine instantiations with create_drm calls for DRM object creation
- Added conditional debug logs for unavailable audio and subtitle tracks when hydrating
- Cleaned up DRM imports and type annotations for drm_system attribute in Netflix class
This commit is contained in:
kenzuyaa
2025-09-09 18:03:14 +07:00
parent 5d4b71b388
commit 6a15cd0a5d
1 changed files with 51 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
unshackle/services/Netflix/__init__.py
View File

@@ -18,12 +18,13 @@ from Crypto.Random import get_random_bytes
import jsonpickle
from pymp4.parser import Box
from pywidevine import PSSH, Cdm, DeviceTypes
from pyplayready import PSSH as PlayReadyPSSH
import requests
from langcodes import Language
from unshackle.core.constants import AnyTrack
from unshackle.core.credential import Credential
from unshackle.core.drm.widevine import Widevine
from unshackle.core.drm import Widevine, PlayReady
from unshackle.core.service import Service
from unshackle.core.titles import Titles_T, Title_T
from unshackle.core.titles.episode import Episode, Series
@@ -36,6 +37,7 @@ from unshackle.core.tracks.subtitle import Subtitle
from unshackle.core.tracks.track import Track
from unshackle.core.tracks.video import Video
from unshackle.core.utils.collections import flatten, as_list
from unshackle.core.drm import DRM_T
from unshackle.core.tracks.attachment import Attachment
from unshackle.core.drm.playready import PlayReady
@@ -83,7 +85,7 @@ class Netflix(Service):
self.profile = profile
self.meta_lang = meta_lang
self.hydrate_track = hydrate_track
self.drm_system = drm_system
self.drm_system: Literal["widevine", "playready"] = drm_system
self.profiles: List[str] = []
self.requested_profiles: List[str] = []
self.high_bitrate = high_bitrate
@@ -248,7 +250,10 @@ class Netflix(Service):
self.log.debug(f"Range {range_index + 1}/{len(self.range)} ({video_range.name}), Profile Index: {profile_index}. Getting profiles: {profile_list}")
manifest = self.get_manifest(title, profile_list)
manifest_tracks = self.manifest_as_tracks(manifest, title, should_hydrate and profile_index == 0)
tracks.add(manifest_tracks if should_hydrate and profile_index == 0 else manifest_tracks.videos)
if should_hydrate and profile_index == 0:
tracks.add(manifest_tracks) # Add all tracks (video, audio, subtitles) on first hydrated profile
else:
tracks.add(manifest_tracks.videos) # Add only videos for additional profiles
except Exception:
self.log.error(f"Error getting profile: {profile_list} for range {video_range.name}. Skipping")
continue
@@ -262,7 +267,12 @@ class Netflix(Service):
self.log.info(f"Processing range {range_index + 1}/{len(self.range)}: {video_range.name}")
manifest = self.get_manifest(title, range_profiles)
manifest_tracks = self.manifest_as_tracks(manifest, title, should_hydrate)
tracks.add(manifest_tracks if should_hydrate else manifest_tracks.videos)
if should_hydrate:
tracks.add(manifest_tracks) # Add all tracks (video, audio, subtitles) when hydrating
elif range_index == 0:
tracks.add(manifest_tracks) # Add all tracks on first range even without hydration
else:
tracks.add(manifest_tracks.videos) # Add only videos for additional ranges
except Exception as e:
self.log.error(f"Error processing range {video_range.name}: {e}")
@@ -392,17 +402,14 @@ class Netflix(Service):
return payload_data[0]["licenseResponseBase64"]
def get_playready_license(self, *, challenge: bytes, title: Movie | Episode | Song, track: AnyTrack) -> bytes | str | None:
return None
return self.get_widevine_license(challenge=challenge, title=title, track=track)
# return super().get_widevine_license(challenge=challenge, title=title, track=track)
def configure(self):
# self.log.info(ctx)
# if profile is none from argument let's use them all profile in video codec scope
# self.log.info(f"Requested profiles: {self.profile}")
if self.profile is None:
self.profiles = self.config["profiles"]["video"][self.vcodec.extension.upper()]
if self.profile is not None:
self.requested_profiles = self.profile.split('+')
self.log.info(f"Requested profile: {self.requested_profiles}")
@@ -430,7 +437,7 @@ class Netflix(Service):
sys.exit(1)
self.profiles = self.get_profiles()
self.drm_system = self.get_drm_system()
# Log information about video ranges being processed
if len(self.range) > 1:
range_names = [r.name for r in self.range]
@@ -808,6 +815,10 @@ class Netflix(Service):
# self.log.info(video)
id = video["downloadable_id"]
# self.log.info(f"Adding video {video["res_w"]}x{video["res_h"]}, bitrate: {(float(video["framerate_value"]) / video["framerate_scale"]) if "framerate_value" in video else None} with profile {video["content_profile"]}. kid: {video["drmHeaderId"]}")
drm = Widevine(
pssh=PSSH(manifest["video_tracks"][0]["drmHeader"]["bytes"]),
pssh_b64=video["drmHeaderId"],
)
tracks.add(
Video(
id_=video["downloadable_id"],
@@ -821,12 +832,7 @@ class Netflix(Service):
edition=video["content_profile"],
range_=self.parse_video_range_from_profile(video["content_profile"]),
is_original_lang=True,
drm=[Widevine(
pssh=PSSH(
manifest["video_tracks"][0]["drmHeader"]["bytes"]
),
kid=video["drmHeaderId"]
)] if manifest["video_tracks"][0].get("drmHeader", {}).get("bytes") else [],
drm=[self.create_drm(manifest["video_tracks"][0]["drmHeader"]["bytes"], video["drmHeaderId"])] if manifest["video_tracks"][0].get("drmHeader", {}).get("bytes") else [],
data={
'license_url': license_url
}
@@ -850,7 +856,8 @@ class Netflix(Service):
# self.log.debug(f"Audio lang {audio["languageDescription"]} is available but no stream available.")
if "new_track_id" in audio and "id" in audio:
unavailable_audio_tracks.append((audio["new_track_id"], audio["id"])) # Assign to `unavailable_subtitle` for request missing audio tracks later
self.log.debug(f"Audio track at index {audio_index}, audio_id: {audio_id}, language: {audio_lang} has no streams available")
if hydrate_tracks:
self.log.debug(f"Audio track at index {audio_index}, audio_id: {audio_id}, language: {audio_lang} has no streams available")
continue
# Store primary audio track info (new_track_id, id) for potential use in hydration
@@ -903,8 +910,8 @@ class Netflix(Service):
# This subtitles is there but has to request stream first
if "new_track_id" in subtitle and "id" in subtitle:
unavailable_subtitle.append((subtitle["new_track_id"], subtitle["id"])) # Assign to `unavailable_subtitle` for request missing subtitles later
# self.log.debug(f"Audio language: {subtitle["languageDescription"]} id: {subtitle["new_track_id"]} is not hydrated.")
self.log.debug(f"Subtitle track at index {subtitle_index}, subtitle_id: {subtitle_id}, language: {subtitle_lang} is not hydrated")
if hydrate_tracks:
self.log.debug(f"Subtitle track at index {subtitle_index}, subtitle_id: {subtitle_id}, language: {subtitle_lang} is not hydrated")
continue
if subtitle.get("languageDescription") == 'Off' and self.descriptive_subtitles == False:
@@ -1189,6 +1196,31 @@ class Netflix(Service):
return hydrated_tracks
def create_drm(self, pssh: str, kid: str) -> DRM_T:
if self.drm_system == "widevine":
return Widevine(PSSH(pssh), kid)
elif self.drm_system == "playready":
return PlayReady(PlayReadyPSSH(pssh), kid, pssh)
else:
raise ValueError("Unknown DRM system while creating DRM")
def get_drm_system(self) -> Literal["widevine", "playready"]:
# This is widevine?
if isinstance(self.cdm, Widevine):
return "widevine"
elif isinstance(self.cdm, PlayReady):
return "playready"
else:
# Maybe this is DecryptLabsRemoteCDM
from unshackle.core.cdm import DecryptLabsRemoteCDM
if (isinstance(self.cdm, DecryptLabsRemoteCDM)):
# Is Decrypt Labs using PlayReady?
if self.cdm.is_playready:
return "playready"
else:
return "widevine"
raise ValueError("Unknown DRM system")
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."""