9 Commits

11 changed files with 835 additions and 331 deletions

View File

@@ -5,6 +5,27 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
- **Custom Output Templates**: Flexible filename customization system
- New `output_template` configuration in unshackle.yaml for movies, series, and songs
- Support for conditional variables using `?` suffix (e.g., `{year?}`, `{hdr?}`)
- Comprehensive template variables for title, quality, audio, video, and metadata
- Multiple naming styles: Scene-style (dot-separated), Plex-friendly (space-separated), minimal, custom
- Automatic template validation and enhanced error handling
- **Full backward compatibility**: Old `scene_naming` option still works and automatically converts to equivalent templates
- Folder naming now follows series template patterns (excluding episode-specific variables)
- Deprecation warnings guide users to migrate from `scene_naming` to `output_template`
### Changed
- **Filename Generation**: Updated all title classes (Movie, Episode, Song) to use new template system
- Enhanced context building for template variable substitution
- Improved separator handling based on template style detection
- Better handling of conditional content like HDR, Atmos, and multi-language audio
## [1.4.4] - 2025-09-02
### Added

View File

@@ -70,10 +70,24 @@ class DecryptLabsRemoteCDMExceptions:
class DecryptLabsRemoteCDM:
"""
Decrypt Labs Remote CDM implementation compatible with pywidevine's CDM interface.
Decrypt Labs Remote CDM implementation with intelligent caching system.
This class provides a drop-in replacement for pywidevine's local CDM using
Decrypt Labs' KeyXtractor API service.
Decrypt Labs' KeyXtractor API service, enhanced with smart caching logic
that minimizes unnecessary license requests.
Key Features:
- Compatible with both Widevine and PlayReady DRM schemes
- Intelligent caching that compares required vs. available keys
- Automatic key combination for mixed cache/license scenarios
- Seamless fallback to license requests when keys are missing
Intelligent Caching System:
1. DRM classes (PlayReady/Widevine) provide required KIDs via set_required_kids()
2. get_license_challenge() first checks for cached keys
3. If cached keys satisfy requirements, returns empty challenge (no license needed)
4. If keys are missing, makes targeted license request for remaining keys
5. parse_license() combines cached and license keys intelligently
"""
service_certificate_challenge = b"\x08\x04"
@@ -127,6 +141,7 @@ class DecryptLabsRemoteCDM:
self._sessions: Dict[bytes, Dict[str, Any]] = {}
self._pssh_b64 = None
self._required_kids: Optional[List[str]] = None
self._http_session = Session()
self._http_session.headers.update(
{
@@ -160,6 +175,29 @@ class DecryptLabsRemoteCDM:
"""Store base64-encoded PSSH data for PlayReady compatibility."""
self._pssh_b64 = pssh_b64
def set_required_kids(self, kids: List[Union[str, UUID]]) -> None:
"""
Set the required Key IDs for intelligent caching decisions.
This method enables the CDM to make smart decisions about when to request
additional keys via license challenges. When cached keys are available,
the CDM will compare them against the required KIDs to determine if a
license request is still needed for missing keys.
Args:
kids: List of required Key IDs as UUIDs or hex strings
Note:
Should be called by DRM classes (PlayReady/Widevine) before making
license challenge requests to enable optimal caching behavior.
"""
self._required_kids = []
for kid in kids:
if isinstance(kid, UUID):
self._required_kids.append(str(kid).replace("-", "").lower())
else:
self._required_kids.append(str(kid).replace("-", "").lower())
def _generate_session_id(self) -> bytes:
"""Generate a unique session ID."""
return secrets.token_bytes(16)
@@ -292,34 +330,25 @@ class DecryptLabsRemoteCDM:
raise DecryptLabsRemoteCDMExceptions.InvalidSession(f"Invalid session ID: {session_id.hex()}")
session = self._sessions[session_id]
pssh = session.get("pssh")
if not pssh:
return False
if self.vaults:
key_ids = []
if hasattr(pssh, "key_ids"):
key_ids = pssh.key_ids
elif hasattr(pssh, "kids"):
key_ids = pssh.kids
for kid in key_ids:
key, _ = self.vaults.get_key(kid)
if key and key.count("0") != len(key):
return True
session_keys = session.get("keys", [])
if session_keys and len(session_keys) > 0:
return True
return False
return len(session_keys) > 0
def get_license_challenge(
self, session_id: bytes, pssh_or_wrm: Any, license_type: str = "STREAMING", privacy_mode: bool = True
) -> bytes:
"""
Generate a license challenge using Decrypt Labs API.
Generate a license challenge using Decrypt Labs API with intelligent caching.
This method implements smart caching logic that:
1. First attempts to retrieve cached keys from the API
2. If required KIDs are set, compares cached keys against requirements
3. Only makes a license request if keys are missing
4. Returns empty challenge if all required keys are cached
The intelligent caching works as follows:
- With required KIDs set: Only requests license for missing keys
- Without required KIDs: Returns any available cached keys
- For PlayReady: Combines cached keys with license keys seamlessly
Args:
session_id: Session identifier
@@ -328,11 +357,14 @@ class DecryptLabsRemoteCDM:
privacy_mode: Whether to use privacy mode - for compatibility only
Returns:
License challenge as bytes
License challenge as bytes, or empty bytes if cached keys satisfy requirements
Raises:
InvalidSession: If session ID is invalid
requests.RequestException: If API request fails
Note:
Call set_required_kids() before this method for optimal caching behavior.
"""
_ = license_type, privacy_mode
@@ -343,8 +375,13 @@ class DecryptLabsRemoteCDM:
session["pssh"] = pssh_or_wrm
init_data = self._get_init_data_from_pssh(pssh_or_wrm)
already_tried_cache = session.get("tried_cache", False)
request_data = {"scheme": self.device_name, "init_data": init_data, "get_cached_keys_if_exists": True}
request_data = {
"scheme": self.device_name,
"init_data": init_data,
"get_cached_keys_if_exists": not already_tried_cache,
}
if self.device_name in ["L1", "L2", "SL2", "SL3"] and self.service_name:
request_data["service"] = self.service_name
@@ -365,22 +402,86 @@ class DecryptLabsRemoteCDM:
error_msg += f" - Details: {data['details']}"
if "error" in data:
error_msg += f" - Error: {data['error']}"
if "service_certificate is required" in str(data) and not session["service_certificate"]:
error_msg += " (No service certificate was provided to the CDM session)"
raise requests.RequestException(f"API error: {error_msg}")
if data.get("message_type") == "cached-keys" or "cached_keys" in data:
cached_keys = data.get("cached_keys", [])
session["keys"] = self._parse_cached_keys(cached_keys)
return b""
message_type = data.get("message_type")
if message_type == "cached-keys" or "cached_keys" in data:
"""
Handle cached keys response from API.
When the API returns cached keys, we need to determine if they satisfy
our requirements or if we need to make an additional license request
for missing keys.
"""
cached_keys = data.get("cached_keys", [])
parsed_keys = self._parse_cached_keys(cached_keys)
session["keys"] = parsed_keys
session["tried_cache"] = True
if self._required_kids:
cached_kids = set()
for key in parsed_keys:
if isinstance(key, dict) and "kid" in key:
cached_kids.add(key["kid"].replace("-", "").lower())
required_kids = set(self._required_kids)
missing_kids = required_kids - cached_kids
if missing_kids:
session["cached_keys"] = parsed_keys
request_data["get_cached_keys_if_exists"] = False
response = self._http_session.post(f"{self.host}/get-request", json=request_data, timeout=30)
if response.status_code == 200:
data = response.json()
if data.get("message") == "success" and "challenge" in data:
challenge = base64.b64decode(data["challenge"])
session["challenge"] = challenge
session["decrypt_labs_session_id"] = data["session_id"]
return challenge
return b""
else:
return b""
else:
return b""
if message_type == "license-request" or "challenge" in data:
challenge = base64.b64decode(data["challenge"])
session["challenge"] = challenge
session["decrypt_labs_session_id"] = data["session_id"]
return challenge
error_msg = f"Unexpected API response format. message_type={message_type}, available_fields={list(data.keys())}"
if data.get("message"):
error_msg = f"API response: {data['message']} - {error_msg}"
if "details" in data:
error_msg += f" - Details: {data['details']}"
if "error" in data:
error_msg += f" - Error: {data['error']}"
if already_tried_cache and data.get("message") == "success":
return b""
raise requests.RequestException(error_msg)
def parse_license(self, session_id: bytes, license_message: Union[bytes, str]) -> None:
"""
Parse license response using Decrypt Labs API.
Parse license response using Decrypt Labs API with intelligent key combination.
For PlayReady content with partial cached keys, this method intelligently
combines the cached keys with newly obtained license keys, avoiding
duplicates while ensuring all required keys are available.
The key combination process:
1. Extracts keys from the license response
2. If cached keys exist (PlayReady), combines them with license keys
3. Removes duplicate keys by comparing normalized KIDs
4. Updates the session with the complete key set
Args:
session_id: Session identifier
@@ -395,7 +496,7 @@ class DecryptLabsRemoteCDM:
session = self._sessions[session_id]
if session["keys"]:
if session["keys"] and not (self.is_playready and "cached_keys" in session):
return
if not session.get("challenge") or not session.get("decrypt_labs_session_id"):
@@ -439,7 +540,48 @@ class DecryptLabsRemoteCDM:
error_msg += f" - Details: {data['details']}"
raise requests.RequestException(f"License decrypt error: {error_msg}")
session["keys"] = self._parse_keys_response(data)
license_keys = self._parse_keys_response(data)
if self.is_playready and "cached_keys" in session:
"""
Combine cached keys with license keys for PlayReady content.
This ensures we have both the cached keys (obtained earlier) and
any additional keys from the license response, without duplicates.
"""
cached_keys = session.get("cached_keys", [])
all_keys = list(cached_keys)
for license_key in license_keys:
already_exists = False
license_kid = None
if isinstance(license_key, dict) and "kid" in license_key:
license_kid = license_key["kid"].replace("-", "").lower()
elif hasattr(license_key, "kid"):
license_kid = str(license_key.kid).replace("-", "").lower()
elif hasattr(license_key, "key_id"):
license_kid = str(license_key.key_id).replace("-", "").lower()
if license_kid:
for cached_key in cached_keys:
cached_kid = None
if isinstance(cached_key, dict) and "kid" in cached_key:
cached_kid = cached_key["kid"].replace("-", "").lower()
elif hasattr(cached_key, "kid"):
cached_kid = str(cached_key.kid).replace("-", "").lower()
elif hasattr(cached_key, "key_id"):
cached_kid = str(cached_key.key_id).replace("-", "").lower()
if cached_kid == license_kid:
already_exists = True
break
if not already_exists:
all_keys.append(license_key)
session["keys"] = all_keys
else:
session["keys"] = license_keys
if self.vaults and session["keys"]:
key_dict = {UUID(hex=key["kid"]): key["key"] for key in session["keys"] if key["type"] == "CONTENT"}
@@ -470,26 +612,6 @@ class DecryptLabsRemoteCDM:
return keys
def _load_cached_keys(self, session_id: bytes) -> None:
"""Load cached keys from vaults and Decrypt Labs API."""
session = self._sessions[session_id]
pssh = session["pssh"]
keys = []
if self.vaults:
key_ids = []
if hasattr(pssh, "key_ids"):
key_ids = pssh.key_ids
elif hasattr(pssh, "kids"):
key_ids = pssh.kids
for kid in key_ids:
key, _ = self.vaults.get_key(kid)
if key and key.count("0") != len(key):
keys.append({"kid": kid.hex, "key": key, "type": "CONTENT"})
session["keys"] = keys
def _parse_cached_keys(self, cached_keys_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Parse cached keys from API response.

View File

@@ -1,5 +1,7 @@
from __future__ import annotations
import re
import warnings
from pathlib import Path
from typing import Any, Optional
@@ -90,13 +92,116 @@ class Config:
self.tmdb_api_key: str = kwargs.get("tmdb_api_key") or ""
self.update_checks: bool = kwargs.get("update_checks", True)
self.update_check_interval: int = kwargs.get("update_check_interval", 24)
self.scene_naming: bool = kwargs.get("scene_naming", True)
self.series_year: bool = kwargs.get("series_year", True)
# Handle backward compatibility for scene_naming option
self.scene_naming: Optional[bool] = kwargs.get("scene_naming")
self.output_template: dict = kwargs.get("output_template") or {}
# Apply scene_naming compatibility if no output_template is defined
self._apply_scene_naming_compatibility()
# Validate output templates
self._validate_output_templates()
self.title_cache_time: int = kwargs.get("title_cache_time", 1800) # 30 minutes default
self.title_cache_max_retention: int = kwargs.get("title_cache_max_retention", 86400) # 24 hours default
self.title_cache_enabled: bool = kwargs.get("title_cache_enabled", True)
def _apply_scene_naming_compatibility(self) -> None:
"""Apply backward compatibility for the old scene_naming option."""
if self.scene_naming is not None:
# Only apply if no output_template is already defined
if not self.output_template.get("movies") and not self.output_template.get("series"):
if self.scene_naming:
# scene_naming: true = scene-style templates
self.output_template.update(
{
"movies": "{title}.{year}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}",
"series": "{title}.{year?}.{season_episode}.{episode_name?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}",
"songs": "{track_number}.{title}.{source?}.WEB-DL.{audio_full}.{atmos?}-{tag}",
}
)
else:
# scene_naming: false = Plex-friendly templates
self.output_template.update(
{
"movies": "{title} ({year}) {quality}",
"series": "{title} {season_episode} {episode_name?}",
"songs": "{track_number}. {title}",
}
)
# Warn about deprecated option
warnings.warn(
"The 'scene_naming' option is deprecated. Please use 'output_template' instead. "
"Your current setting has been converted to equivalent templates.",
DeprecationWarning,
stacklevel=2,
)
def _validate_output_templates(self) -> None:
"""Validate output template configurations and warn about potential issues."""
if not self.output_template:
return
# Known template variables for validation
valid_variables = {
# Basic variables
"title",
"year",
"season",
"episode",
"season_episode",
"episode_name",
"quality",
"resolution",
"source",
"tag",
"track_number",
"artist",
"album",
"disc",
# Audio variables
"audio",
"audio_channels",
"audio_full",
"atmos",
"dual",
"multi",
# Video variables
"video",
"hdr",
"hfr",
}
# Filesystem-unsafe characters that could cause issues
unsafe_chars = r'[<>:"/\\|?*]'
for template_type, template_str in self.output_template.items():
if not isinstance(template_str, str):
warnings.warn(f"Template '{template_type}' must be a string, got {type(template_str).__name__}")
continue
# Extract variables from template
variables = re.findall(r"\{([^}]+)\}", template_str)
# Check for unknown variables
for var in variables:
# Remove conditional suffix if present
var_clean = var.rstrip("?")
if var_clean not in valid_variables:
warnings.warn(f"Unknown template variable '{var}' in {template_type} template")
# Check for filesystem-unsafe characters outside of variables
# Replace variables with safe placeholders for testing
test_template = re.sub(r"\{[^}]+\}", "TEST", template_str)
if re.search(unsafe_chars, test_template):
warnings.warn(f"Template '{template_type}' may contain filesystem-unsafe characters")
# Check for empty template
if not template_str.strip():
warnings.warn(f"Template '{template_type}' is empty")
@classmethod
def from_yaml(cls, path: Path) -> Config:
if not path.exists():

View File

@@ -256,28 +256,19 @@ class PlayReady:
return keys
def get_content_keys(self, cdm: PlayReadyCdm, certificate: Callable, licence: Callable) -> None:
for kid in self.kids:
if kid in self.content_keys:
continue
session_id = cdm.open()
try:
if hasattr(cdm, "set_pssh_b64") and self.pssh_b64:
cdm.set_pssh_b64(self.pssh_b64)
if hasattr(cdm, "set_required_kids"):
cdm.set_required_kids(self.kids)
challenge = cdm.get_license_challenge(session_id, self.pssh.wrm_headers[0])
if challenge:
try:
license_res = licence(challenge=challenge)
except Exception:
if hasattr(cdm, "use_cached_keys_as_fallback"):
if cdm.use_cached_keys_as_fallback(session_id):
keys = self._extract_keys_from_cdm(cdm, session_id)
self.content_keys.update(keys)
continue
raise
if isinstance(license_res, bytes):
license_str = license_res.decode(errors="ignore")
else:
@@ -290,6 +281,9 @@ class PlayReady:
pass
cdm.parse_license(session_id, license_str)
except Exception:
raise
keys = self._extract_keys_from_cdm(cdm, session_id)
self.content_keys.update(keys)
finally:

View File

@@ -185,6 +185,9 @@ class Widevine:
if cert and hasattr(cdm, "set_service_certificate"):
cdm.set_service_certificate(session_id, cert)
if hasattr(cdm, "set_required_kids"):
cdm.set_required_kids(self.kids)
challenge = cdm.get_license_challenge(session_id, self.pssh)
if hasattr(cdm, "has_cached_keys") and cdm.has_cached_keys(session_id):
@@ -218,6 +221,9 @@ class Widevine:
if cert and hasattr(cdm, "set_service_certificate"):
cdm.set_service_certificate(session_id, cert)
if hasattr(cdm, "set_required_kids"):
cdm.set_required_kids(self.kids)
challenge = cdm.get_license_challenge(session_id, self.pssh)
if hasattr(cdm, "has_cached_keys") and cdm.has_cached_keys(session_id):

View File

@@ -12,6 +12,7 @@ from unshackle.core.config import config
from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP
from unshackle.core.titles.title import Title
from unshackle.core.utilities import sanitize_filename
from unshackle.core.utils.template_formatter import TemplateFormatter
class Episode(Title):
@@ -78,117 +79,155 @@ class Episode(Title):
self.year = year
self.description = description
def _build_template_context(self, media_info: MediaInfo, show_service: bool = True) -> dict:
"""Build template context dictionary from MediaInfo."""
primary_video_track = next(iter(media_info.video_tracks), None)
primary_audio_track = next(iter(media_info.audio_tracks), None)
unique_audio_languages = len({x.language.split("-")[0] for x in media_info.audio_tracks if x.language})
context = {
"title": self.title.replace("$", "S"),
"year": self.year or "",
"season": f"S{self.season:02}",
"episode": f"E{self.number:02}",
"season_episode": f"S{self.season:02}E{self.number:02}",
"episode_name": self.name or "",
"tag": config.tag or "",
"source": self.service.__name__ if show_service else "",
}
# Video information
if primary_video_track:
resolution = primary_video_track.height
aspect_ratio = [int(float(plane)) for plane in primary_video_track.other_display_aspect_ratio[0].split(":")]
if len(aspect_ratio) == 1:
aspect_ratio.append(1)
if aspect_ratio[0] / aspect_ratio[1] not in (16 / 9, 4 / 3):
resolution = int(primary_video_track.width * (9 / 16))
context.update(
{
"quality": f"{resolution}p",
"resolution": str(resolution),
"video": VIDEO_CODEC_MAP.get(primary_video_track.format, primary_video_track.format),
}
)
# HDR information
hdr_format = primary_video_track.hdr_format_commercial
trc = primary_video_track.transfer_characteristics or primary_video_track.transfer_characteristics_original
if hdr_format:
if (primary_video_track.hdr_format or "").startswith("Dolby Vision"):
context["hdr"] = "DV"
base_layer = DYNAMIC_RANGE_MAP.get(hdr_format)
if base_layer and base_layer != "DV":
context["hdr"] += f".{base_layer}"
else:
context["hdr"] = DYNAMIC_RANGE_MAP.get(hdr_format, "")
elif trc and "HLG" in trc:
context["hdr"] = "HLG"
else:
context["hdr"] = ""
# High frame rate
frame_rate = float(primary_video_track.frame_rate)
context["hfr"] = "HFR" if frame_rate > 30 else ""
# Audio information
if primary_audio_track:
codec = primary_audio_track.format
channel_layout = primary_audio_track.channel_layout or primary_audio_track.channellayout_original
if channel_layout:
channels = float(sum({"LFE": 0.1}.get(position.upper(), 1) for position in channel_layout.split(" ")))
else:
channel_count = primary_audio_track.channel_s or primary_audio_track.channels or 0
channels = float(channel_count)
features = primary_audio_track.format_additionalfeatures or ""
context.update(
{
"audio": AUDIO_CODEC_MAP.get(codec, codec),
"audio_channels": f"{channels:.1f}",
"audio_full": f"{AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}",
"atmos": "Atmos" if ("JOC" in features or primary_audio_track.joc) else "",
}
)
# Multi-language audio
if unique_audio_languages == 2:
context["dual"] = "DUAL"
context["multi"] = ""
elif unique_audio_languages > 2:
context["dual"] = ""
context["multi"] = "MULTi"
else:
context["dual"] = ""
context["multi"] = ""
return context
def __str__(self) -> str:
return "{title}{year} S{season:02}E{number:02} {name}".format(
title=self.title,
year=f" {self.year}" if self.year and config.series_year else "",
year=f" {self.year}" if self.year else "",
season=self.season,
number=self.number,
name=self.name or "",
).strip()
def get_filename(self, media_info: MediaInfo, folder: bool = False, show_service: bool = True) -> str:
primary_video_track = next(iter(media_info.video_tracks), None)
primary_audio_track = next(iter(media_info.audio_tracks), None)
unique_audio_languages = len({x.language.split("-")[0] for x in media_info.audio_tracks if x.language})
# Title [Year] SXXEXX Name (or Title [Year] SXX if folder)
if folder:
# For folders, use the series template but exclude episode-specific variables
series_template = config.output_template.get("series")
if series_template:
# Create a folder-friendly version by removing episode-specific variables
folder_template = series_template
# Remove episode number and episode name from template for folders
folder_template = re.sub(r'\{episode\}', '', folder_template)
folder_template = re.sub(r'\{episode_name\?\}', '', folder_template)
folder_template = re.sub(r'\{episode_name\}', '', folder_template)
folder_template = re.sub(r'\{season_episode\}', '{season}', folder_template)
# Clean up any double separators that might result
folder_template = re.sub(r'\.{2,}', '.', folder_template)
folder_template = re.sub(r'\s{2,}', ' ', folder_template)
folder_template = re.sub(r'^[\.\s]+|[\.\s]+$', '', folder_template)
formatter = TemplateFormatter(folder_template)
context = self._build_template_context(media_info, show_service)
# Override season_episode with just season for folders
context['season'] = f"S{self.season:02}"
folder_name = formatter.format(context)
# Keep the same separator style as the series template
if '.' in series_template and ' ' not in series_template:
# Dot-based template - use dot separator for folders too
return sanitize_filename(folder_name, ".")
else:
# Space-based template - use space separator
return sanitize_filename(folder_name, " ")
else:
# Fallback to simple naming if no template defined
name = f"{self.title}"
if self.year and config.series_year:
if self.year:
name += f" {self.year}"
name += f" S{self.season:02}"
else:
name = "{title}{year} S{season:02}E{number:02} {name}".format(
title=self.title.replace("$", "S"), # e.g., Arli$$
year=f" {self.year}" if self.year and config.series_year else "",
season=self.season,
number=self.number,
name=self.name or "",
).strip()
if config.scene_naming:
# Resolution
if primary_video_track:
resolution = primary_video_track.height
aspect_ratio = [
int(float(plane)) for plane in primary_video_track.other_display_aspect_ratio[0].split(":")
]
if len(aspect_ratio) == 1:
# e.g., aspect ratio of 2 (2.00:1) would end up as `(2.0,)`, add 1
aspect_ratio.append(1)
if aspect_ratio[0] / aspect_ratio[1] not in (16 / 9, 4 / 3):
# We want the resolution represented in a 4:3 or 16:9 canvas.
# If it's not 4:3 or 16:9, calculate as if it's inside a 16:9 canvas,
# otherwise the track's height value is fine.
# We are assuming this title is some weird aspect ratio so most
# likely a movie or HD source, so it's most likely widescreen so
# 16:9 canvas makes the most sense.
resolution = int(primary_video_track.width * (9 / 16))
name += f" {resolution}p"
# Service
if show_service:
name += f" {self.service.__name__}"
# 'WEB-DL'
name += " WEB-DL"
# DUAL
if unique_audio_languages == 2:
name += " DUAL"
# MULTi
if unique_audio_languages > 2:
name += " MULTi"
# Audio Codec + Channels (+ feature)
if primary_audio_track:
codec = primary_audio_track.format
channel_layout = primary_audio_track.channel_layout or primary_audio_track.channellayout_original
if channel_layout:
channels = float(
sum({"LFE": 0.1}.get(position.upper(), 1) for position in channel_layout.split(" "))
)
else:
channel_count = primary_audio_track.channel_s or primary_audio_track.channels or 0
channels = float(channel_count)
features = primary_audio_track.format_additionalfeatures or ""
name += f" {AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}"
if "JOC" in features or primary_audio_track.joc:
name += " Atmos"
# Video (dynamic range + hfr +) Codec
if primary_video_track:
codec = primary_video_track.format
hdr_format = primary_video_track.hdr_format_commercial
trc = (
primary_video_track.transfer_characteristics
or primary_video_track.transfer_characteristics_original
)
frame_rate = float(primary_video_track.frame_rate)
if hdr_format:
if (primary_video_track.hdr_format or "").startswith("Dolby Vision"):
name += " DV"
if DYNAMIC_RANGE_MAP.get(hdr_format) and DYNAMIC_RANGE_MAP.get(hdr_format) != "DV":
name += " HDR"
else:
name += f" {DYNAMIC_RANGE_MAP.get(hdr_format)} "
elif trc and "HLG" in trc:
name += " HLG"
if frame_rate > 30:
name += " HFR"
name += f" {VIDEO_CODEC_MAP.get(codec, codec)}"
if config.tag:
name += f"-{config.tag}"
return sanitize_filename(name)
else:
# Simple naming style without technical details - use spaces instead of dots
return sanitize_filename(name, " ")
# Use template from output_template (which includes scene_naming compatibility)
# or fallback to default scene-style template
template = (
config.output_template.get("series")
or "{title}.{year?}.{season_episode}.{episode_name?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hfr?}.{video}-{tag}"
)
formatter = TemplateFormatter(template)
context = self._build_template_context(media_info, show_service)
return formatter.format(context)
class Series(SortedKeyList, ABC):
def __init__(self, iterable: Optional[Iterable] = None):
@@ -197,7 +236,7 @@ class Series(SortedKeyList, ABC):
def __str__(self) -> str:
if not self:
return super().__str__()
return self[0].title + (f" ({self[0].year})" if self[0].year and config.series_year else "")
return self[0].title + (f" ({self[0].year})" if self[0].year else "")
def tree(self, verbose: bool = False) -> Tree:
seasons = Counter(x.season for x in self)

View File

@@ -9,7 +9,7 @@ from sortedcontainers import SortedKeyList
from unshackle.core.config import config
from unshackle.core.constants import AUDIO_CODEC_MAP, DYNAMIC_RANGE_MAP, VIDEO_CODEC_MAP
from unshackle.core.titles.title import Title
from unshackle.core.utilities import sanitize_filename
from unshackle.core.utils.template_formatter import TemplateFormatter
class Movie(Title):
@@ -45,100 +45,107 @@ class Movie(Title):
self.year = year
self.description = description
def _build_template_context(self, media_info: MediaInfo, show_service: bool = True) -> dict:
"""Build template context dictionary from MediaInfo."""
primary_video_track = next(iter(media_info.video_tracks), None)
primary_audio_track = next(iter(media_info.audio_tracks), None)
unique_audio_languages = len({x.language.split("-")[0] for x in media_info.audio_tracks if x.language})
context = {
"title": self.name.replace("$", "S"),
"year": self.year or "",
"tag": config.tag or "",
"source": self.service.__name__ if show_service else "",
}
# Video information
if primary_video_track:
resolution = primary_video_track.height
aspect_ratio = [int(float(plane)) for plane in primary_video_track.other_display_aspect_ratio[0].split(":")]
if len(aspect_ratio) == 1:
aspect_ratio.append(1)
if aspect_ratio[0] / aspect_ratio[1] not in (16 / 9, 4 / 3):
resolution = int(primary_video_track.width * (9 / 16))
context.update(
{
"quality": f"{resolution}p",
"resolution": str(resolution),
"video": VIDEO_CODEC_MAP.get(primary_video_track.format, primary_video_track.format),
}
)
# HDR information
hdr_format = primary_video_track.hdr_format_commercial
trc = primary_video_track.transfer_characteristics or primary_video_track.transfer_characteristics_original
if hdr_format:
if (primary_video_track.hdr_format or "").startswith("Dolby Vision"):
context["hdr"] = "DV"
base_layer = DYNAMIC_RANGE_MAP.get(hdr_format)
if base_layer and base_layer != "DV":
context["hdr"] += f".{base_layer}"
else:
context["hdr"] = DYNAMIC_RANGE_MAP.get(hdr_format, "")
elif trc and "HLG" in trc:
context["hdr"] = "HLG"
else:
context["hdr"] = ""
# High frame rate
frame_rate = float(primary_video_track.frame_rate)
context["hfr"] = "HFR" if frame_rate > 30 else ""
# Audio information
if primary_audio_track:
codec = primary_audio_track.format
channel_layout = primary_audio_track.channel_layout or primary_audio_track.channellayout_original
if channel_layout:
channels = float(sum({"LFE": 0.1}.get(position.upper(), 1) for position in channel_layout.split(" ")))
else:
channel_count = primary_audio_track.channel_s or primary_audio_track.channels or 0
channels = float(channel_count)
features = primary_audio_track.format_additionalfeatures or ""
context.update(
{
"audio": AUDIO_CODEC_MAP.get(codec, codec),
"audio_channels": f"{channels:.1f}",
"audio_full": f"{AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}",
"atmos": "Atmos" if ("JOC" in features or primary_audio_track.joc) else "",
}
)
# Multi-language audio
if unique_audio_languages == 2:
context["dual"] = "DUAL"
context["multi"] = ""
elif unique_audio_languages > 2:
context["dual"] = ""
context["multi"] = "MULTi"
else:
context["dual"] = ""
context["multi"] = ""
return context
def __str__(self) -> str:
if self.year:
return f"{self.name} ({self.year})"
return self.name
def get_filename(self, media_info: MediaInfo, folder: bool = False, show_service: bool = True) -> str:
primary_video_track = next(iter(media_info.video_tracks), None)
primary_audio_track = next(iter(media_info.audio_tracks), None)
unique_audio_languages = len({x.language.split("-")[0] for x in media_info.audio_tracks if x.language})
# Name (Year)
name = str(self).replace("$", "S") # e.g., Arli$$
if config.scene_naming:
# Resolution
if primary_video_track:
resolution = primary_video_track.height
aspect_ratio = [
int(float(plane)) for plane in primary_video_track.other_display_aspect_ratio[0].split(":")
]
if len(aspect_ratio) == 1:
# e.g., aspect ratio of 2 (2.00:1) would end up as `(2.0,)`, add 1
aspect_ratio.append(1)
if aspect_ratio[0] / aspect_ratio[1] not in (16 / 9, 4 / 3):
# We want the resolution represented in a 4:3 or 16:9 canvas.
# If it's not 4:3 or 16:9, calculate as if it's inside a 16:9 canvas,
# otherwise the track's height value is fine.
# We are assuming this title is some weird aspect ratio so most
# likely a movie or HD source, so it's most likely widescreen so
# 16:9 canvas makes the most sense.
resolution = int(primary_video_track.width * (9 / 16))
name += f" {resolution}p"
# Service
if show_service:
name += f" {self.service.__name__}"
# 'WEB-DL'
name += " WEB-DL"
# DUAL
if unique_audio_languages == 2:
name += " DUAL"
# MULTi
if unique_audio_languages > 2:
name += " MULTi"
# Audio Codec + Channels (+ feature)
if primary_audio_track:
codec = primary_audio_track.format
channel_layout = primary_audio_track.channel_layout or primary_audio_track.channellayout_original
if channel_layout:
channels = float(
sum({"LFE": 0.1}.get(position.upper(), 1) for position in channel_layout.split(" "))
# Use template from output_template (which includes scene_naming compatibility)
# or fallback to default scene-style template
template = (
config.output_template.get("movies")
or "{title}.{year}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
)
else:
channel_count = primary_audio_track.channel_s or primary_audio_track.channels or 0
channels = float(channel_count)
features = primary_audio_track.format_additionalfeatures or ""
name += f" {AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}"
if "JOC" in features or primary_audio_track.joc:
name += " Atmos"
# Video (dynamic range + hfr +) Codec
if primary_video_track:
codec = primary_video_track.format
hdr_format = primary_video_track.hdr_format_commercial
trc = (
primary_video_track.transfer_characteristics
or primary_video_track.transfer_characteristics_original
)
frame_rate = float(primary_video_track.frame_rate)
if hdr_format:
if (primary_video_track.hdr_format or "").startswith("Dolby Vision"):
name += " DV"
if DYNAMIC_RANGE_MAP.get(hdr_format) and DYNAMIC_RANGE_MAP.get(hdr_format) != "DV":
name += " HDR"
else:
name += f" {DYNAMIC_RANGE_MAP.get(hdr_format)} "
elif trc and "HLG" in trc:
name += " HLG"
if frame_rate > 30:
name += " HFR"
name += f" {VIDEO_CODEC_MAP.get(codec, codec)}"
if config.tag:
name += f"-{config.tag}"
return sanitize_filename(name)
else:
# Simple naming style without technical details - use spaces instead of dots
return sanitize_filename(name, " ")
formatter = TemplateFormatter(template)
context = self._build_template_context(media_info, show_service)
return formatter.format(context)
class Movies(SortedKeyList, ABC):

View File

@@ -10,6 +10,7 @@ from unshackle.core.config import config
from unshackle.core.constants import AUDIO_CODEC_MAP
from unshackle.core.titles.title import Title
from unshackle.core.utilities import sanitize_filename
from unshackle.core.utils.template_formatter import TemplateFormatter
class Song(Title):
@@ -81,46 +82,63 @@ class Song(Title):
artist=self.artist, album=self.album, year=self.year, track=self.track, name=self.name
).strip()
def get_filename(self, media_info: MediaInfo, folder: bool = False, show_service: bool = True) -> str:
audio_track = next(iter(media_info.audio_tracks), None)
codec = audio_track.format
channel_layout = audio_track.channel_layout or audio_track.channellayout_original
def _build_template_context(self, media_info: MediaInfo, show_service: bool = True) -> dict:
"""Build template context dictionary from MediaInfo."""
primary_audio_track = next(iter(media_info.audio_tracks), None)
context = {
"artist": self.artist.replace("$", "S"),
"album": self.album.replace("$", "S"),
"title": self.name.replace("$", "S"),
"track_number": f"{self.track:02}",
"disc": f"{self.disc:02}" if self.disc > 1 else "",
"year": self.year or "",
"tag": config.tag or "",
"source": self.service.__name__ if show_service else "",
}
# Audio information
if primary_audio_track:
codec = primary_audio_track.format
channel_layout = primary_audio_track.channel_layout or primary_audio_track.channellayout_original
if channel_layout:
channels = float(sum({"LFE": 0.1}.get(position.upper(), 1) for position in channel_layout.split(" ")))
else:
channel_count = audio_track.channel_s or audio_track.channels or 0
channel_count = primary_audio_track.channel_s or primary_audio_track.channels or 0
channels = float(channel_count)
features = audio_track.format_additionalfeatures or ""
features = primary_audio_track.format_additionalfeatures or ""
context.update(
{
"audio": AUDIO_CODEC_MAP.get(codec, codec),
"audio_channels": f"{channels:.1f}",
"audio_full": f"{AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}",
"atmos": "Atmos" if ("JOC" in features or primary_audio_track.joc) else "",
}
)
return context
def get_filename(self, media_info: MediaInfo, folder: bool = False, show_service: bool = True) -> str:
if folder:
# Artist - Album (Year)
name = str(self).split(" / ")[0]
else:
# NN. Song Name
name = str(self).split(" / ")[1]
if config.scene_naming:
# Service
if show_service:
name += f" {self.service.__name__}"
# 'WEB-DL'
name += " WEB-DL"
# Audio Codec + Channels (+ feature)
name += f" {AUDIO_CODEC_MAP.get(codec, codec)}{channels:.1f}"
if "JOC" in features or audio_track.joc:
name += " Atmos"
if config.tag:
name += f"-{config.tag}"
return sanitize_filename(name, " ")
else:
# Simple naming style without technical details
# For folders, use simple naming: "Artist - Album (Year)"
name = f"{self.artist} - {self.album}"
if self.year:
name += f" ({self.year})"
return sanitize_filename(name, " ")
# Use template from output_template (which includes scene_naming compatibility)
# or fallback to default scene-style template
template = (
config.output_template.get("songs") or "{track_number}.{title}.{source?}.WEB-DL.{audio_full}.{atmos?}-{tag}"
)
formatter = TemplateFormatter(template)
context = self._build_template_context(media_info, show_service)
return formatter.format(context)
class Album(SortedKeyList, ABC):
def __init__(self, iterable: Optional[Iterable] = None):

View File

@@ -350,14 +350,26 @@ def tag_file(path: Path, title: Title, tmdb_id: Optional[int] | None = None) ->
if simkl_tmdb_id:
tmdb_id = simkl_tmdb_id
# Handle TV show data from Simkl
if simkl_data.get("type") == "episode" and "show" in simkl_data:
show_ids = simkl_data.get("show", {}).get("ids", {})
if show_ids.get("imdb"):
standard_tags["IMDB"] = show_ids["imdb"]
if show_ids.get("tvdb"):
standard_tags["TVDB"] = str(show_ids["tvdb"])
standard_tags["TVDB2"] = f"series/{show_ids['tvdb']}"
if show_ids.get("tmdbtv"):
standard_tags["TMDB"] = f"tv/{show_ids['tmdbtv']}"
# Handle movie data from Simkl
elif simkl_data.get("type") == "movie" and "movie" in simkl_data:
movie_ids = simkl_data.get("movie", {}).get("ids", {})
if movie_ids.get("imdb"):
standard_tags["IMDB"] = movie_ids["imdb"]
if movie_ids.get("tvdb"):
standard_tags["TVDB2"] = f"movies/{movie_ids['tvdb']}"
if movie_ids.get("tmdb"):
standard_tags["TMDB"] = f"movie/{movie_ids['tmdb']}"
# Use TMDB API for additional metadata (either from provided ID or Simkl lookup)
api_key = _api_key()
if not api_key:
@@ -389,7 +401,10 @@ def tag_file(path: Path, title: Title, tmdb_id: Optional[int] | None = None) ->
standard_tags["IMDB"] = imdb_id
tvdb_id = ids.get("tvdb_id")
if tvdb_id:
standard_tags["TVDB"] = str(tvdb_id)
if kind == "movie":
standard_tags["TVDB2"] = f"movies/{tvdb_id}"
else:
standard_tags["TVDB2"] = f"series/{tvdb_id}"
merged_tags = {
**custom_tags,

View File

@@ -0,0 +1,147 @@
import logging
import re
from typing import Any, Dict, List
from unshackle.core.utilities import sanitize_filename
class TemplateFormatter:
"""
Template formatter for custom filename patterns.
Supports variable substitution and conditional variables.
Example: '{title}.{year}.{quality?}.{source}-{tag}'
"""
def __init__(self, template: str):
"""Initialize the template formatter.
Args:
template: Template string with variables in {variable} format
"""
self.template = template
self.variables = self._extract_variables()
def _extract_variables(self) -> List[str]:
"""Extract all variables from the template."""
pattern = r"\{([^}]+)\}"
matches = re.findall(pattern, self.template)
return [match.strip() for match in matches]
def format(self, context: Dict[str, Any]) -> str:
"""Format the template with the provided context.
Args:
context: Dictionary containing variable values
Returns:
Formatted filename string
Raises:
ValueError: If required template variables are missing from context
"""
logger = logging.getLogger(__name__)
# Validate that all required variables are present
is_valid, missing_vars = self.validate(context)
if not is_valid:
error_msg = f"Missing required template variables: {', '.join(missing_vars)}"
logger.error(error_msg)
raise ValueError(error_msg)
try:
result = self.template
for variable in self.variables:
placeholder = "{" + variable + "}"
is_conditional = variable.endswith("?")
if is_conditional:
# Remove the ? for conditional variables
var_name = variable[:-1]
value = context.get(var_name, "")
if value:
# Replace with actual value, ensuring it's string and safe
safe_value = str(value).strip()
result = result.replace(placeholder, safe_value)
else:
# Remove the placeholder entirely for empty conditional variables
result = result.replace(placeholder, "")
else:
# Regular variable
value = context.get(variable, "")
if value is None:
logger.warning(f"Template variable '{variable}' is None, using empty string")
value = ""
safe_value = str(value).strip()
result = result.replace(placeholder, safe_value)
# Clean up multiple consecutive dots/separators and other artifacts
result = re.sub(r"\.{2,}", ".", result) # Multiple dots -> single dot
result = re.sub(r"\s{2,}", " ", result) # Multiple spaces -> single space
result = re.sub(r"^[\.\s]+|[\.\s]+$", "", result) # Remove leading/trailing dots and spaces
result = re.sub(r"\.-", "-", result) # Remove dots before dashes (for dot-based templates)
result = re.sub(r"[\.\s]+\)", ")", result) # Remove dots/spaces before closing parentheses
# Determine the appropriate separator based on template style
# If the template contains spaces (like Plex-friendly), preserve them
if " " in self.template and "." not in self.template:
# Space-based template (Plex-friendly) - use space separator
result = sanitize_filename(result, spacer=" ")
else:
# Dot-based template (scene-style) - use dot separator
result = sanitize_filename(result, spacer=".")
# Final validation - ensure we have a non-empty result
if not result or result.isspace():
logger.warning("Template formatting resulted in empty filename, using fallback")
return "untitled"
logger.debug(f"Template formatted successfully: '{self.template}' -> '{result}'")
return result
except Exception as e:
logger.error(f"Error formatting template '{self.template}': {e}")
# Return a safe fallback filename
fallback = f"error_formatting_{hash(self.template) % 10000}"
logger.warning(f"Using fallback filename: {fallback}")
return fallback
def validate(self, context: Dict[str, Any]) -> tuple[bool, List[str]]:
"""Validate that all required variables are present in context.
Args:
context: Dictionary containing variable values
Returns:
Tuple of (is_valid, missing_variables)
"""
missing = []
for variable in self.variables:
is_conditional = variable.endswith("?")
var_name = variable[:-1] if is_conditional else variable
# Only check non-conditional variables
if not is_conditional and var_name not in context:
missing.append(var_name)
return len(missing) == 0, missing
def get_required_variables(self) -> List[str]:
"""Get list of required (non-conditional) variables."""
required = []
for variable in self.variables:
if not variable.endswith("?"):
required.append(variable)
return required
def get_optional_variables(self) -> List[str]:
"""Get list of optional (conditional) variables."""
optional = []
for variable in self.variables:
if variable.endswith("?"):
optional.append(variable[:-1]) # Remove the ?
return optional

View File

@@ -10,15 +10,45 @@ tag_imdb_tmdb: true
# Set terminal background color (custom option not in CONFIG.md)
set_terminal_bg: false
# Set file naming convention
# true for style - Prime.Suspect.S07E01.The.Final.Act.Part.One.1080p.ITV.WEB-DL.AAC2.0.H.264
# false for style - Prime Suspect S07E01 The Final Act - Part One
scene_naming: true
# File naming is now controlled via output_template (see below)
# Default behavior provides scene-style naming similar to the old scene_naming: true
#
# BACKWARD COMPATIBILITY: The old scene_naming option is still supported:
# scene_naming: true -> Equivalent to scene-style templates (dot-separated)
# scene_naming: false -> Equivalent to Plex-friendly templates (space-separated)
# Note: output_template takes precedence over scene_naming if both are defined
# Whether to include the year in series names for episodes and folders (default: true)
# true for style - Show Name (2023) S01E01 Episode Name
# false for style - Show Name S01E01 Episode Name
series_year: true
# Custom output templates for filenames
# When not defined, defaults to scene-style naming equivalent to the old scene_naming: true
# Available variables: {title}, {year}, {season}, {episode}, {season_episode}, {episode_name},
# {quality}, {resolution}, {source}, {audio}, {audio_channels}, {audio_full},
# {video}, {hdr}, {hfr}, {atmos}, {dual}, {multi}, {tag}
# Conditional variables (included only if present): Add ? suffix like {year?}, {episode_name?}, {hdr?}
# Uncomment and customize the templates below:
#
# output_template:
# # Scene-style naming (dot-separated) - Default behavior when no template is defined
# movies: '{title}.{year}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}'
# series: '{title}.{year?}.{season_episode}.{episode_name?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}'
#
# # Plex-friendly naming (space-separated, clean format)
# # movies: '{title} ({year}) {quality}'
# # series: '{title} {season_episode} {episode_name?}'
#
# # Minimal naming (basic info only)
# # movies: '{title}.{year}.{quality}'
# # series: '{title}.{season_episode}.{episode_name?}'
#
# # Custom scene-style with specific elements
# # movies: '{title}.{year}.{quality}.{hdr?}.{source}.WEB-DL.{audio_full}.{video}-{tag}'
# # series: '{title}.{year?}.{season_episode}.{episode_name?}.{quality}.{hdr?}.{source}.WEB-DL.{audio_full}.{atmos?}.{video}-{tag}'
#
# Example outputs:
# Scene movies: 'The.Matrix.1999.1080p.NF.WEB-DL.DDP5.1.H.264-EXAMPLE'
# Scene movies (HDR): 'Dune.2021.2160p.HBO.WEB-DL.DDP5.1.HDR10.H.265-EXAMPLE'
# Scene series: 'Breaking.Bad.2008.S01E01.Pilot.1080p.NF.WEB-DL.DDP5.1.H.264-EXAMPLE'
# Plex movies: 'The Matrix (1999) 1080p'
# Plex series: 'Breaking Bad S01E01 Pilot'
# Check for updates from GitHub repository on startup (default: true)
update_checks: true