sefree/unshackle-SeFree
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'dev' into service.py

Browse Source
This commit is contained in:
CodeName393
2026-02-25 19:21:22 +09:00
committed by GitHub
parent 86d464dc8e 21f0186ebf
commit 05dc682a2d
19 changed files with 625 additions and 1606 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
unshackle/core/__init__.py
View File

@@ -1 +1 @@
__version__ = "3.0.0"
__version__ = "3.1.0"

104
unshackle/core/service.py
View File

@@ -1,7 +1,8 @@
import base64
import logging
from abc import ABCMeta, abstractmethod
from collections.abc import Generator
from collections.abc import Callable, Generator
from dataclasses import dataclass, field
from http.cookiejar import CookieJar
from pathlib import Path
from typing import Optional, Union
@@ -24,9 +25,26 @@ from unshackle.core.search_result import SearchResult
from unshackle.core.title_cacher import TitleCacher, get_account_hash, get_region_from_proxy
from unshackle.core.titles import Title_T, Titles_T
from unshackle.core.tracks import Chapters, Tracks
from unshackle.core.tracks.video import Video
from unshackle.core.utilities import get_cached_ip_info, get_ip_info
@dataclass
class TrackRequest:
"""Holds what the user requested for video codec and range selection.
Services read from this instead of ctx.parent.params for vcodec/range.
Attributes:
codecs: Requested codecs from CLI. Empty list means no filter (accept any).
ranges: Requested ranges from CLI. Defaults to [SDR].
"""
codecs: list[Video.Codec] = field(default_factory=list)
ranges: list[Video.Range] = field(default_factory=lambda: [Video.Range.SDR])
best_available: bool = False
def sanitize_proxy_for_log(uri: Optional[str]) -> Optional[str]:
"""
Sanitize a proxy URI for logs by redacting any embedded userinfo (username/password).
@@ -89,6 +107,16 @@ class Service(metaclass=ABCMeta):
self.credential = None # Will be set in authenticate()
self.current_region = None # Will be set based on proxy/geolocation
# Set track request from CLI params - services can read/override in their __init__
vcodec = ctx.parent.params.get("vcodec") if ctx.parent else None
range_ = ctx.parent.params.get("range_") if ctx.parent else None
best_available = ctx.parent.params.get("best_available", False) if ctx.parent else False
self.track_request = TrackRequest(
codecs=list(vcodec) if vcodec else [],
ranges=list(range_) if range_ else [Video.Range.SDR],
best_available=bool(best_available),
)
if not ctx.parent or not ctx.parent.params.get("no_proxy"):
if ctx.parent:
proxy = ctx.parent.params["proxy"]
@@ -205,6 +233,76 @@ class Service(metaclass=ABCMeta):
self.log.debug(f"Failed to get cached IP info: {e}")
self.current_region = None
def _get_tracks_for_variants(
self,
title: Title_T,
fetch_fn: Callable[..., Tracks],
) -> Tracks:
"""Call fetch_fn for each codec/range combo in track_request, merge results.
Services that need separate API calls per codec/range combo can use this
helper from their get_tracks() implementation.
The fetch_fn signature should be: (title, codec, range_) -> Tracks
For HYBRID range, fetch_fn is called with HDR10 and DV separately and
the DV video tracks are merged into the HDR10 result.
Args:
title: The title being processed.
fetch_fn: A callable that fetches tracks for a specific codec/range.
"""
all_tracks = Tracks()
first = True
codecs = self.track_request.codecs or [None]
ranges = self.track_request.ranges or [Video.Range.SDR]
for range_val in ranges:
if range_val == Video.Range.HYBRID:
# HYBRID: fetch HDR10 first (full tracks), then DV (video only)
for codec_val in codecs:
try:
hdr_tracks = fetch_fn(title, codec=codec_val, range_=Video.Range.HDR10)
except (ValueError, SystemExit) as e:
if self.track_request.best_available:
self.log.warning(f" - HDR10 not available for HYBRID, skipping ({e})")
continue
raise
if first:
all_tracks.add(hdr_tracks, warn_only=True)
first = False
else:
for video in hdr_tracks.videos:
all_tracks.add(video, warn_only=True)
try:
dv_tracks = fetch_fn(title, codec=codec_val, range_=Video.Range.DV)
for video in dv_tracks.videos:
all_tracks.add(video, warn_only=True)
except (ValueError, SystemExit):
self.log.info(" - No DolbyVision manifest available for HYBRID")
else:
for codec_val in codecs:
try:
tracks = fetch_fn(title, codec=codec_val, range_=range_val)
except (ValueError, SystemExit) as e:
if self.track_request.best_available:
codec_name = codec_val.name if codec_val else "default"
self.log.warning(
f" - {range_val.name}/{codec_name} not available, skipping ({e})"
)
continue
raise
if first:
all_tracks.add(tracks, warn_only=True)
first = False
else:
for video in tracks.videos:
all_tracks.add(video, warn_only=True)
return all_tracks
# Optional Abstract functions
# The following functions may be implemented by the Service.
# Otherwise, the base service code (if any) of the function will be executed on call.
@@ -222,7 +320,7 @@ class Service(metaclass=ABCMeta):
session.mount(
"https://",
HTTPAdapter(
max_retries=Retry(total=15, backoff_factor=0.2, status_forcelist=[429, 500, 502, 503, 504]),
max_retries=Retry(total=5, backoff_factor=0.2, status_forcelist=[429, 500, 502, 503, 504]),
pool_block=True,
),
)
@@ -482,4 +580,4 @@ class Service(metaclass=ABCMeta):
"""
__all__ = ("Service",)
__all__ = ("Service", "TrackRequest")

6
unshackle/core/session.py
View File

@@ -56,7 +56,7 @@ class MaxRetriesError(exceptions.RequestException):
class CurlSession(Session):
def __init__(
self,
max_retries: int = 10,
max_retries: int = 5,
backoff_factor: float = 0.2,
max_backoff: float = 60.0,
status_forcelist: list[int] | None = None,
@@ -150,7 +150,7 @@ def session(
browser: Browser to impersonate (e.g. "chrome124", "firefox", "safari") OR
fingerprint preset name (e.g. "okhttp4").
Uses the configured default from curl_impersonate.browser if not specified.
Available presets: okhttp4
Available presets: okhttp4, okhttp5
See https://github.com/lexiforest/curl_cffi#sessions for browser options.
ja3: Custom JA3 TLS fingerprint string (format: "SSLVersion,Ciphers,Extensions,Curves,PointFormats").
When provided, curl_cffi will use this exact TLS fingerprint instead of the browser's default.
@@ -172,7 +172,7 @@ def session(
- cert: Client certificate (str or tuple)
Extra arguments for retry handler:
- max_retries: Maximum number of retries (int, default 10)
- max_retries: Maximum number of retries (int, default 5)
- backoff_factor: Backoff factor (float, default 0.2)
- max_backoff: Maximum backoff time (float, default 60.0)
- status_forcelist: List of status codes to force retry (list, default [429, 500, 502, 503, 504])

25
unshackle/core/tracks/track.py
View File

@@ -24,7 +24,7 @@ from unshackle.core.constants import DOWNLOAD_CANCELLED, DOWNLOAD_LICENCE_ONLY
from unshackle.core.downloaders import aria2c, curl_impersonate, n_m3u8dl_re, requests
from unshackle.core.drm import DRM_T, PlayReady, Widevine
from unshackle.core.events import events
from unshackle.core.utilities import get_boxes, get_extension, try_ensure_utf8
from unshackle.core.utilities import get_boxes, try_ensure_utf8
from unshackle.core.utils.subprocess import ffprobe
@@ -210,23 +210,12 @@ class Track:
save_path = config.directories.temp / f"{track_type}_{self.id}.mp4"
if track_type == "Subtitle":
save_path = save_path.with_suffix(f".{self.codec.extension}")
# n_m3u8dl_re doesn't support directly downloading subtitles from URLs
# or when the subtitle has a direct file extension
if self.downloader.__name__ == "n_m3u8dl_re" and (
self.descriptor == self.Descriptor.URL
or get_extension(self.url)
in {
".srt",
".vtt",
".ttml",
".ssa",
".ass",
".stpp",
".wvtt",
".xml",
}
):
self.downloader = requests
if self.downloader.__name__ == "n_m3u8dl_re" and (
self.descriptor == self.Descriptor.URL
or track_type in ("Subtitle", "Attachment")
):
self.downloader = requests
if self.descriptor != self.Descriptor.URL:
save_dir = save_path.with_name(save_path.name + "_segments")

27
unshackle/core/utils/click_types.py
View File

@@ -44,6 +44,33 @@ class VideoCodecChoice(click.Choice):
self.fail(f"'{value}' is not a valid video codec", param, ctx)
class MultipleVideoCodecChoice(VideoCodecChoice):
"""
A multiple-value variant of VideoCodecChoice that accepts comma-separated codecs.
Accepts both enum names and values, e.g.: ``-v hevc,avc`` or ``-v H.264,H.265``
"""
name = "multiple_video_codec_choice"
def convert(
self, value: Any, param: Optional[click.Parameter] = None, ctx: Optional[click.Context] = None
) -> list[Any]:
if not value:
return []
if isinstance(value, list):
values = value
elif isinstance(value, str):
values = value.split(",")
else:
self.fail(f"{value!r} is not a supported value.", param, ctx)
chosen_values: list[Any] = []
for v in values:
chosen_values.append(super().convert(v.strip(), param, ctx))
return chosen_values
class SubtitleCodecChoice(click.Choice):
"""
A custom Choice type for subtitle codecs that accepts both enum names, values, and common aliases.