mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-15 13:35:05 +00:00
imSp4rky
ecdfb667c9
--sample downloads only a time window of a title (default 120-180 s) and cuts the mux to it. --include-sample cuts a 60 s sample from a finished file. Both write to a Sample folder with a SAMPLE tag on the filename
1046 lines
44 KiB
Python
1046 lines
44 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from functools import partial
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterator, Optional, Sequence, Union
|
|
|
|
from langcodes import Language
|
|
from rich.progress import Progress, SpinnerColumn, TaskID, TextColumn, TimeRemainingColumn
|
|
from rich.table import Table
|
|
from rich.tree import Tree
|
|
|
|
from unshackle.core import binaries
|
|
from unshackle.core.config import config
|
|
from unshackle.core.console import GradientPulseBarColumn, console
|
|
from unshackle.core.constants import AnyTrack, TrackT
|
|
from unshackle.core.events import events
|
|
from unshackle.core.tracks.attachment import Attachment
|
|
from unshackle.core.tracks.audio import Audio
|
|
from unshackle.core.tracks.chapters import Chapter, Chapters
|
|
from unshackle.core.tracks.subtitle import Subtitle
|
|
from unshackle.core.tracks.track import Track, has_dts_uhd_sample_entry, strip_duplicate_init_boxes
|
|
from unshackle.core.tracks.video import Video
|
|
from unshackle.core.utilities import (
|
|
is_close_match,
|
|
log_event,
|
|
matching_languages,
|
|
sanitize_filename,
|
|
valid_language,
|
|
)
|
|
from unshackle.core.utils.collections import as_list, flatten
|
|
from unshackle.core.utils.subprocess import run_step
|
|
|
|
MP4BOX_PROGRESS = re.compile(r"\((\d{1,3})/100\)")
|
|
MP4BOX_OPEN_FAILED = re.compile(r"error while opening|Error opening file|Invalid IsoMedia File", re.I)
|
|
|
|
|
|
class Tracks:
|
|
"""
|
|
Video, Audio, Subtitle, Chapter, and Attachment Track Store.
|
|
It provides convenience functions for listing, sorting, and selecting tracks.
|
|
"""
|
|
|
|
TRACK_ORDER_MAP = {Video: 0, Audio: 1, Subtitle: 2, Chapter: 3, Attachment: 4}
|
|
|
|
def __init__(
|
|
self,
|
|
*args: Union[
|
|
Tracks, Sequence[Union[AnyTrack, Chapter, Chapters, Attachment]], Track, Chapter, Chapters, Attachment
|
|
],
|
|
manifest_url: Optional[str] = None,
|
|
):
|
|
self.videos: list[Video] = []
|
|
self.audio: list[Audio] = []
|
|
self.subtitles: list[Subtitle] = []
|
|
self.chapters = Chapters()
|
|
self.attachments: list[Attachment] = []
|
|
self.manifest_url: Optional[str] = manifest_url
|
|
|
|
if args:
|
|
self.add(args)
|
|
|
|
def __iter__(self) -> Iterator[AnyTrack]:
|
|
return iter(as_list(self.videos, self.audio, self.subtitles))
|
|
|
|
def __len__(self) -> int:
|
|
return len(self.videos) + len(self.audio) + len(self.subtitles)
|
|
|
|
def __add__(
|
|
self,
|
|
other: Union[
|
|
Tracks, Sequence[Union[AnyTrack, Chapter, Chapters, Attachment]], Track, Chapter, Chapters, Attachment
|
|
],
|
|
) -> Tracks:
|
|
self.add(other)
|
|
return self
|
|
|
|
def __repr__(self) -> str:
|
|
return "{name}({items})".format(
|
|
name=self.__class__.__name__, items=", ".join([f"{k}={repr(v)}" for k, v in self.__dict__.items()])
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
rep = {Video: [], Audio: [], Subtitle: [], Chapter: [], Attachment: []}
|
|
tracks = [*list(self), *self.chapters]
|
|
|
|
for track in sorted(tracks, key=lambda t: self.TRACK_ORDER_MAP[type(t)]):
|
|
if not rep[type(track)]:
|
|
count = sum(type(x) is type(track) for x in tracks)
|
|
rep[type(track)].append(
|
|
"{count} {type} Track{plural}{colon}".format(
|
|
count=count,
|
|
type=track.__class__.__name__,
|
|
plural="s" if count != 1 else "",
|
|
colon=":" if count > 0 else "",
|
|
)
|
|
)
|
|
rep[type(track)].append(str(track))
|
|
|
|
for type_ in list(rep):
|
|
if not rep[type_]:
|
|
del rep[type_]
|
|
continue
|
|
rep[type_] = "\n".join([rep[type_][0]] + [f"├─ {x}" for x in rep[type_][1:-1]] + [f"└─ {rep[type_][-1]}"])
|
|
rep = "\n".join(list(rep.values()))
|
|
|
|
return rep
|
|
|
|
def tree(self, add_progress: bool = False) -> tuple[Tree, list[Callable[..., None]]]:
|
|
all_tracks = [*list(self), *self.chapters, *self.attachments]
|
|
|
|
progress_callables = []
|
|
|
|
tree = Tree("", hide_root=True)
|
|
for track_type in self.TRACK_ORDER_MAP:
|
|
tracks = list(x for x in all_tracks if isinstance(x, track_type))
|
|
if tracks:
|
|
num_tracks = len(tracks)
|
|
track_type_plural = track_type.__name__ + ("s" if track_type != Audio and num_tracks != 1 else "")
|
|
tracks_tree = tree.add(f"[repr.number]{num_tracks}[/] {track_type_plural}")
|
|
for track in tracks:
|
|
if add_progress and track_type not in (Chapter, Attachment):
|
|
progress = Progress(
|
|
SpinnerColumn(finished_text=""),
|
|
GradientPulseBarColumn(),
|
|
"•",
|
|
TimeRemainingColumn(compact=True, elapsed_when_finished=True),
|
|
"•",
|
|
TextColumn("[progress.data.speed]{task.fields[downloaded]}"),
|
|
console=console,
|
|
speed_estimate_period=10,
|
|
)
|
|
task = progress.add_task("", downloaded="-")
|
|
state = {"total": 100.0}
|
|
|
|
def update_track_progress(
|
|
task_id: TaskID = task,
|
|
_state: dict[str, float] = state,
|
|
_progress: Progress = progress,
|
|
**kwargs: Any,
|
|
) -> None:
|
|
"""
|
|
Make sure that terminal status states render as a fully completed bar.
|
|
|
|
Some downloaders can report completed slightly below total
|
|
before emitting the final "Downloaded" state.
|
|
"""
|
|
if "total" in kwargs:
|
|
if kwargs["total"] is None:
|
|
# Progress.update() ignores total=None; an un-started task pulses
|
|
del kwargs["total"]
|
|
_progress.reset(task_id, start=False)
|
|
else:
|
|
_state["total"] = kwargs["total"]
|
|
_progress.start_task(task_id)
|
|
|
|
downloaded_state = kwargs.get("downloaded")
|
|
if downloaded_state in {"Downloaded", "Decrypted", "[yellow]SKIPPED"}:
|
|
kwargs["completed"] = _state["total"]
|
|
kwargs["total"] = _state["total"]
|
|
_progress.start_task(task_id)
|
|
_progress.update(task_id=task_id, **kwargs)
|
|
|
|
progress_callables.append(update_track_progress)
|
|
track_table = Table.grid()
|
|
track_table.add_row(str(track)[6:], style="text2")
|
|
track_table.add_row(progress)
|
|
tracks_tree.add(track_table)
|
|
else:
|
|
tracks_tree.add(str(track)[6:], style="text2")
|
|
|
|
# Show Closed Captions right after Subtitles (even if no subtitle tracks exist)
|
|
if track_type is Subtitle:
|
|
seen_cc: set[str] = set()
|
|
unique_cc: list[str] = []
|
|
for video in (x for x in all_tracks if isinstance(x, Video)):
|
|
for cc in getattr(video, "closed_captions", []):
|
|
lang = cc.get("language", "und")
|
|
name = cc.get("name", "")
|
|
instream_id = cc.get("instream_id", "")
|
|
key = f"{lang}|{instream_id}"
|
|
if key in seen_cc:
|
|
continue
|
|
seen_cc.add(key)
|
|
parts = [f"[CC] | {lang}"]
|
|
if name:
|
|
parts.append(name)
|
|
if instream_id:
|
|
parts.append(instream_id)
|
|
unique_cc.append(" | ".join(parts))
|
|
if unique_cc:
|
|
cc_tree = tree.add(
|
|
f"[repr.number]{len(unique_cc)}[/] Closed Caption{'s' if len(unique_cc) != 1 else ''}"
|
|
)
|
|
for cc_str in unique_cc:
|
|
cc_tree.add(cc_str, style="text2")
|
|
|
|
return tree, progress_callables
|
|
|
|
def exists(self, by_id: Optional[str] = None, by_url: Optional[Union[str, list[str]]] = None) -> bool:
|
|
"""Examine whether a track already exists, by various methods."""
|
|
if by_id:
|
|
return any(x.id == by_id for x in self)
|
|
if by_url:
|
|
return any(x.url == by_url for x in self)
|
|
return False
|
|
|
|
def add(
|
|
self,
|
|
tracks: Union[
|
|
Tracks, Sequence[Union[AnyTrack, Chapter, Chapters, Attachment]], Track, Chapter, Chapters, Attachment
|
|
],
|
|
warn_only: bool = False,
|
|
) -> None:
|
|
"""Add a provided track to its appropriate array.
|
|
|
|
A track whose ID is already in the collection raises ValueError. With `warn_only`
|
|
set, this method skips such a track and logs how many it skipped.
|
|
"""
|
|
if isinstance(tracks, Tracks):
|
|
if tracks.manifest_url and not self.manifest_url:
|
|
self.manifest_url = tracks.manifest_url
|
|
tracks = [*list(tracks), *tracks.chapters, *tracks.attachments]
|
|
|
|
duplicates = 0
|
|
for track in flatten(tracks):
|
|
if self.exists(by_id=track.id):
|
|
if not warn_only:
|
|
raise ValueError(
|
|
"One or more of the provided Tracks is a duplicate. "
|
|
"Track IDs must be unique but accurate using static values. The "
|
|
"value should stay the same no matter when you request the same "
|
|
"content. Use a value that has relation to the track content "
|
|
"itself and is static or permanent and not random/RNG data that "
|
|
"wont change each refresh or conflict in edge cases."
|
|
)
|
|
duplicates += 1
|
|
continue
|
|
|
|
if isinstance(track, Video):
|
|
self.videos.append(track)
|
|
elif isinstance(track, Audio):
|
|
self.audio.append(track)
|
|
elif isinstance(track, Subtitle):
|
|
self.subtitles.append(track)
|
|
elif isinstance(track, Chapter):
|
|
self.chapters.add(track)
|
|
elif isinstance(track, Attachment):
|
|
self.attachments.append(track)
|
|
else:
|
|
raise ValueError("Track type was not set or is invalid.")
|
|
|
|
log = logging.getLogger("Tracks")
|
|
|
|
if duplicates:
|
|
log.debug(f" - Found and skipped {duplicates} duplicate tracks...")
|
|
|
|
def sort_videos(
|
|
self, by_language: Optional[Sequence[Union[str, Language]]] = None, exact_match: bool = False
|
|
) -> None:
|
|
"""Sort video tracks by resolution then bitrate, and optionally language."""
|
|
if not self.videos:
|
|
return
|
|
# resolution first, then bitrate (unknown-bitrate tracks still rank by resolution)
|
|
self.videos.sort(key=lambda x: (x.height or 0, float(x.bitrate or 0.0)), reverse=True)
|
|
for language in reversed(by_language or []):
|
|
if str(language) in ("all", "best"):
|
|
language = next((x.language for x in self.videos if x.is_original_lang), "")
|
|
if not language:
|
|
continue
|
|
wanted = matching_languages(language, [x.language for x in self.videos], exact_match)
|
|
self.videos.sort(key=lambda x: str(x.language))
|
|
self.videos.sort(key=lambda x: str(x.language) not in wanted)
|
|
|
|
def sort_audio(
|
|
self,
|
|
by_language: Optional[Sequence[Union[str, Language]]] = None,
|
|
codec_priority: Optional[Sequence[str]] = None,
|
|
exact_match: bool = False,
|
|
) -> None:
|
|
"""Sort audio tracks by bitrate, codec priority, Atmos, descriptive, and optionally language."""
|
|
if not self.audio:
|
|
return
|
|
# bitrate (highest first)
|
|
self.audio.sort(key=lambda x: float(x.bitrate or 0.0), reverse=True)
|
|
# codec priority (listed codecs ranked in order; unlisted fall to end with bitrate order preserved)
|
|
if codec_priority:
|
|
rank = {str(c).upper(): i for i, c in enumerate(codec_priority)}
|
|
default_rank = len(rank)
|
|
self.audio.sort(key=lambda x: rank.get(x.codec.name if x.codec else "", default_rank))
|
|
# Atmos tracks first (prioritize over higher bitrate non-Atmos)
|
|
self.audio.sort(key=lambda x: not x.atmos)
|
|
self.audio.sort(key=lambda x: x.descriptive)
|
|
for language in reversed(by_language or []):
|
|
if str(language) in ("all", "best"):
|
|
language = next((x.language for x in self.audio if x.is_original_lang), "")
|
|
if not language:
|
|
continue
|
|
wanted = matching_languages(language, [x.language for x in self.audio], exact_match)
|
|
self.audio.sort(key=lambda x: str(x.language) not in wanted)
|
|
|
|
def sort_subtitles(
|
|
self,
|
|
by_language: Optional[Sequence[Union[str, Language]]] = None,
|
|
type_priority: Optional[Sequence[str]] = None,
|
|
group_by: Optional[str] = None,
|
|
exact_match: bool = False,
|
|
) -> None:
|
|
"""
|
|
Sort subtitle tracks by various track attributes to a common P2P standard.
|
|
You may optionally give a sequence of languages to prioritise to the top.
|
|
|
|
Section Order:
|
|
- by_language groups prioritized to top, and ascending alphabetically
|
|
- then rest ascending alphabetically after the prioritized groups
|
|
(Each section ascending alphabetically, but separated)
|
|
|
|
Type Order:
|
|
- Forced
|
|
- Normal
|
|
- Hard of Hearing (SDH/CC)
|
|
(Least to most captions expected in the subtitle)
|
|
|
|
type_priority overrides the Type Order with an explicit ranking of "forced",
|
|
"normal", and "sdh" (cc counts as sdh). Unlisted types fall to the end.
|
|
|
|
group_by sets the major sort order. "type" (default) keeps every forced track together,
|
|
then every normal, then every SDH, each block ascending by language. "language"
|
|
groups by language instead, so Finnish sits next to Finnish SDH, with the Type
|
|
Order applied inside each language.
|
|
|
|
exact_match makes a by_language entry sort only its own tag. By default "en" also
|
|
sorts "en-US" and "en-GB".
|
|
"""
|
|
if not self.subtitles:
|
|
return
|
|
|
|
def by_type() -> None:
|
|
if type_priority:
|
|
rank = {str(t).lower(): i for i, t in enumerate(type_priority)}
|
|
default_rank = len(rank)
|
|
self.subtitles.sort(
|
|
key=lambda x: rank.get(
|
|
"forced" if x.forced else "sdh" if (x.sdh or x.cc) else "normal", default_rank
|
|
)
|
|
)
|
|
else:
|
|
self.subtitles.sort(key=lambda x: x.sdh or x.cc)
|
|
self.subtitles.sort(key=lambda x: x.forced, reverse=True)
|
|
|
|
def by_lang() -> None:
|
|
self.subtitles.sort(key=lambda x: str(x.language))
|
|
|
|
# stable sorts, so the last pass is the major key
|
|
if str(group_by or "type").lower() == "language":
|
|
by_type()
|
|
by_lang()
|
|
else:
|
|
by_lang()
|
|
by_type()
|
|
for language in reversed(by_language or []):
|
|
if str(language) in ("all", "best"):
|
|
language = next((x.language for x in self.subtitles if x.is_original_lang), "")
|
|
if not language:
|
|
continue
|
|
wanted = matching_languages(language, [x.language for x in self.subtitles], exact_match)
|
|
self.subtitles.sort(key=lambda x: str(x.language) in wanted, reverse=True)
|
|
|
|
def select_video(self, x: Callable[[Video], bool]) -> None:
|
|
self.videos = list(filter(x, self.videos))
|
|
|
|
def select_audio(self, x: Callable[[Audio], bool]) -> None:
|
|
self.audio = list(filter(x, self.audio))
|
|
|
|
def select_subtitles(self, x: Callable[[Subtitle], bool]) -> None:
|
|
self.subtitles = list(filter(x, self.subtitles))
|
|
|
|
def filter(self, predicate: Callable[[AnyTrack], bool]) -> Tracks:
|
|
"""Return a new Tracks with tracks filtered by predicate, preserving metadata."""
|
|
new_tracks = Tracks(manifest_url=self.manifest_url)
|
|
new_tracks.videos = [t for t in self.videos if predicate(t)]
|
|
new_tracks.audio = [t for t in self.audio if predicate(t)]
|
|
new_tracks.subtitles = [t for t in self.subtitles if predicate(t)]
|
|
new_tracks.chapters = self.chapters
|
|
new_tracks.attachments = list(self.attachments)
|
|
return new_tracks
|
|
|
|
@staticmethod
|
|
def merge_video_selections(*groups: list[Video]) -> list[Video]:
|
|
"""Concatenate video selections, dropping duplicates (by track id, order-preserving).
|
|
|
|
A caller can choose a DV track as both the hybrid ingredient (lowest) and an
|
|
explicit deliverable. Without dedup, unshackle would mux and download the same
|
|
track twice.
|
|
"""
|
|
merged: list[Video] = []
|
|
for group in groups:
|
|
for video in group:
|
|
if video not in merged:
|
|
merged.append(video)
|
|
return merged
|
|
|
|
@staticmethod
|
|
def partition_hybrid_videos(
|
|
videos: list[Video], non_hybrid_ranges: list[Video.Range]
|
|
) -> tuple[list[Video], list[Video]]:
|
|
"""Split videos into hybrid-ingredient candidates and the standalone-deliverable pool.
|
|
|
|
HDR10/HDR10+/DV tracks are hybrid ingredients. They only enter the standalone
|
|
pool when the user explicitly requested their range alongside HYBRID, so for
|
|
example `-r HYBRID` muxes only the hybrid while `-r HYBRID,HDR10P` also delivers
|
|
HDR10+.
|
|
"""
|
|
ingredient_ranges = (Video.Range.HDR10, Video.Range.HDR10P, Video.Range.DV)
|
|
hybrid_candidates = [v for v in videos if v.range in ingredient_ranges]
|
|
non_hybrid = [v for v in videos if v.range not in ingredient_ranges or v.range in non_hybrid_ranges]
|
|
return hybrid_candidates, non_hybrid
|
|
|
|
@staticmethod
|
|
def flag_hybrid_ingredients(hybrid_selected: list[Video], non_hybrid_selected: list[Video]) -> None:
|
|
"""Mark tracks selected only as hybrid ingredients so the standalone mux loop skips them.
|
|
|
|
A track that the caller also selected as an explicit deliverable (same track in both
|
|
selections) stays unflagged, and the standalone mux loop muxes it alongside the hybrid.
|
|
"""
|
|
for video in hybrid_selected:
|
|
if video not in non_hybrid_selected:
|
|
video.hybrid_base_only = True
|
|
|
|
def select_hybrid(self, tracks, quality, worst: bool = False):
|
|
# Prefer HDR10+ over HDR10 as the base layer (preserves dynamic metadata)
|
|
base_ranges = (Video.Range.HDR10P, Video.Range.HDR10)
|
|
base_tracks = []
|
|
for range_type in base_ranges:
|
|
base_tracks = [
|
|
v
|
|
for v in tracks
|
|
if v.range == range_type and (v.height in quality or (v.width and int(v.width * 9 / 16) in quality))
|
|
]
|
|
if base_tracks:
|
|
break
|
|
|
|
pick = min if worst else max
|
|
base_selected = []
|
|
for res in quality:
|
|
candidates = [v for v in base_tracks if v.height == res or (v.width and int(v.width * 9 / 16) == res)]
|
|
if candidates:
|
|
chosen = pick(candidates, key=lambda v: v.bitrate)
|
|
base_selected.append(chosen)
|
|
|
|
dv_tracks = [v for v in tracks if v.range == Video.Range.DV]
|
|
lowest_dv = min(dv_tracks, key=lambda v: v.height) if dv_tracks else None
|
|
|
|
def select(x):
|
|
if x in base_selected:
|
|
return True
|
|
if lowest_dv and x is lowest_dv:
|
|
return True
|
|
return False
|
|
|
|
return select
|
|
|
|
def by_resolutions(self, resolutions: list[int], per_resolution: int = 0) -> None:
|
|
groups: dict[tuple, list[Video]] = {}
|
|
for video in self.videos:
|
|
groups.setdefault((video.range, video.codec), []).append(video)
|
|
|
|
keep: set[int] = set()
|
|
for resolution in resolutions:
|
|
for group in groups.values():
|
|
matches = [x for x in group if x.height == resolution]
|
|
if not matches:
|
|
matches = [x for x in group if x.width and int(x.width * (9 / 16)) == resolution]
|
|
keep.update(id(x) for x in matches[: per_resolution or None])
|
|
self.videos = [x for x in self.videos if id(x) in keep]
|
|
|
|
@staticmethod
|
|
def by_language(
|
|
tracks: list[TrackT], languages: list[str], per_language: int = 0, exact_match: bool = False
|
|
) -> list[TrackT]:
|
|
selected: list[TrackT] = []
|
|
seen_ids: set[str] = set()
|
|
for language in languages:
|
|
wanted = matching_languages(language, [x.language for x in tracks], exact_match)
|
|
matches = [x for x in tracks if str(x.language) in wanted]
|
|
# Overlapping tags can still resolve to the same physical track; dedupe by id so
|
|
# callers never get duplicate tracks.
|
|
for track in matches[: per_language or None]:
|
|
if track.id not in seen_ids:
|
|
seen_ids.add(track.id)
|
|
selected.append(track)
|
|
return selected
|
|
|
|
@property
|
|
def sample_base(self) -> float:
|
|
"""The earliest ``sample_start`` of the video and audio tracks, which a sample mux uses as time zero."""
|
|
return min((t.sample_start for t in (*self.videos, *self.audio)), default=0.0)
|
|
|
|
def _mux_mp4(
|
|
self,
|
|
title: str,
|
|
delete: bool = True,
|
|
progress: Optional[partial] = None,
|
|
skip_subtitles: bool = False,
|
|
output_path: Optional[Path] = None,
|
|
) -> tuple[Path, int, list[str]]:
|
|
"""
|
|
Multiplex the Tracks into an MP4 Container with MP4Box.
|
|
|
|
Only used for a codec Matroska cannot name, currently DTS-UHD, because MP4 carries
|
|
far less: it drops the forced, hearing-impaired and original flags, holds no
|
|
attachments at all, and leaves out a subtitle it has no format for rather than
|
|
mangling it. Losing those beats shipping audio no player can name.
|
|
|
|
Returns the same (path, returncode, errors) shape as the Matroska mux.
|
|
"""
|
|
mp4box = getattr(binaries, "MP4Box", None)
|
|
if not mp4box:
|
|
raise RuntimeError(
|
|
"MP4Box (GPAC) is required to mux DTS-UHD (DTS:X Profile 2) audio but was not found. "
|
|
"Install it with your package manager (e.g. `apt install gpac`), then check it with "
|
|
"`unshackle env check`."
|
|
)
|
|
|
|
progress = progress or partial(lambda **kwargs: None)
|
|
|
|
cl = [str(mp4box)]
|
|
sample_base = self.sample_base
|
|
names: list[str] = []
|
|
cleaned: list[Path] = []
|
|
track_id = 0
|
|
|
|
for track in [*self.videos, *self.audio, *([] if skip_subtitles else self.subtitles)]:
|
|
if not track.path or not track.path.exists():
|
|
raise ValueError(f"{track.__class__.__name__} Track must be downloaded before muxing...")
|
|
if isinstance(track, Subtitle) and track.path.suffix.lower() in (".ass", ".ssa"):
|
|
log_event(
|
|
"mux_mp4_subtitle_skipped",
|
|
level="WARNING",
|
|
message=f"MP4 cannot store {track.path.suffix.lstrip('.').upper()} subtitles, leaving out {track.language}",
|
|
context={"track_id": track.id, "language": str(track.language), "suffix": track.path.suffix},
|
|
)
|
|
continue
|
|
events.emit(events.Types.TRACK_MULTIPLEX, track=track)
|
|
track_id += 1
|
|
|
|
source = track.path
|
|
candidate = config.directories.temp / f"mp4mux_{track.id}_{os.getpid()}_{threading.get_ident()}.mp4"
|
|
dropped = strip_duplicate_init_boxes(source, candidate)
|
|
if dropped:
|
|
cleaned.append(candidate)
|
|
source = candidate
|
|
log_event(
|
|
"mux_mp4_init_deduplicated",
|
|
level="DEBUG",
|
|
message=f"Dropped {dropped} repeated init box(es) from {track.path.name} for MP4Box",
|
|
context={"track_id": track.id, "dropped": dropped},
|
|
)
|
|
elif candidate.exists():
|
|
candidate.unlink()
|
|
|
|
delay = round((track.sample_start - sample_base) * 1000) if config.sample else 0
|
|
cl.extend(["-add", f"{source}:lang={track.language}" + (f":delay={delay}" if delay else "")])
|
|
name = "" if isinstance(track, Video) else (track.get_track_name() or "")
|
|
if name:
|
|
names.extend(["-name", f"{track_id}={name}"])
|
|
|
|
if not track_id:
|
|
raise ValueError("No tracks provided, at least one track must be provided.")
|
|
|
|
cl.extend(names)
|
|
|
|
chapters_path = None
|
|
if self.chapters and not config.sample:
|
|
chapters_path = config.directories.temp / config.filenames.chapters.format(
|
|
title=sanitize_filename(title), random=f"{self.chapters.id}_{os.getpid()}_{threading.get_ident()}"
|
|
)
|
|
self.chapters.dump(chapters_path, fallback_name=config.chapter_fallback_name)
|
|
cl.extend(["-chap", str(chapters_path)])
|
|
|
|
if self.attachments:
|
|
log_event(
|
|
"mux_mp4_attachments_skipped",
|
|
level="WARNING",
|
|
message=f"MP4 cannot store attachments, leaving out {len(self.attachments)} file(s)",
|
|
context={"count": len(self.attachments)},
|
|
)
|
|
|
|
if output_path is None:
|
|
first = self.videos[0].path if self.videos else self.audio[0].path if self.audio else None
|
|
if first is None:
|
|
raise ValueError("No tracks provided, at least one track must be provided.")
|
|
output_path = first.with_suffix(".muxed.mp4")
|
|
else:
|
|
output_path = output_path.with_suffix(".mp4")
|
|
|
|
full_command = [*cl, "-new", str(output_path)]
|
|
|
|
log_event(
|
|
"mux_start",
|
|
level="INFO",
|
|
message=f"Muxing {len(self.videos)}V/{len(self.audio)}A/{len(self.subtitles)}S -> {output_path.name}",
|
|
context={
|
|
"title": title,
|
|
"output_path": str(output_path),
|
|
"muxer": str(mp4box),
|
|
"command": full_command,
|
|
"container": "mp4",
|
|
"reason": "DTS-UHD audio has no Matroska CodecID",
|
|
},
|
|
)
|
|
|
|
try:
|
|
errors = []
|
|
mux_start_time = time.monotonic()
|
|
p = subprocess.Popen(
|
|
full_command,
|
|
text=True,
|
|
encoding="utf-8",
|
|
errors="replace",
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
assert p.stdout is not None
|
|
for line in iter(p.stdout.readline, ""):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
percent = MP4BOX_PROGRESS.search(line)
|
|
if percent:
|
|
progress(total=100, completed=int(percent.group(1)))
|
|
elif "error" in line.lower() or "not supported" in line.lower():
|
|
errors.append(line)
|
|
|
|
returncode = p.wait()
|
|
if returncode < 2 and any(MP4BOX_OPEN_FAILED.search(line) for line in errors):
|
|
returncode = 2
|
|
if config.sample and returncode == 0:
|
|
start, end = (seconds - sample_base for seconds in config.sample)
|
|
cut_path = output_path.with_suffix(".cut.mp4")
|
|
try:
|
|
run_step(
|
|
[mp4box, "-splitx", f"{start:.3f}:{end:.3f}", output_path, "-out", cut_path],
|
|
output=cut_path,
|
|
label="MP4Box sample cut",
|
|
)
|
|
os.replace(cut_path, output_path)
|
|
except RuntimeError as e:
|
|
returncode = 2
|
|
errors.append(f"MP4Box could not cut the sample window: {e}")
|
|
mux_duration_ms = round((time.monotonic() - mux_start_time) * 1000, 1)
|
|
output_size = output_path.stat().st_size if output_path.exists() else 0
|
|
|
|
log_event(
|
|
"mux_failed" if returncode != 0 else "mux_complete",
|
|
level="ERROR" if returncode != 0 else "INFO",
|
|
message=(
|
|
f"MP4Box exited with code {returncode}"
|
|
if returncode != 0
|
|
else f"Muxed {output_path.name} ({output_size} bytes) in {mux_duration_ms}ms"
|
|
),
|
|
context={
|
|
"output_path": str(output_path),
|
|
"output_size": output_size,
|
|
"duration_ms": mux_duration_ms,
|
|
"returncode": returncode,
|
|
"errors": errors,
|
|
},
|
|
)
|
|
|
|
return output_path, returncode, errors
|
|
finally:
|
|
if chapters_path:
|
|
chapters_path.unlink()
|
|
for path in cleaned:
|
|
path.unlink(missing_ok=True)
|
|
if delete:
|
|
for track in self:
|
|
track.delete()
|
|
|
|
def mux(
|
|
self,
|
|
title: str,
|
|
delete: bool = True,
|
|
progress: Optional[partial] = None,
|
|
audio_expected: bool = True,
|
|
title_language: Optional[Language] = None,
|
|
skip_subtitles: bool = False,
|
|
output_path: Optional[Path] = None,
|
|
) -> tuple[Path, int, list[str]]:
|
|
"""
|
|
Multiplex all the Tracks into a Matroska Container file.
|
|
|
|
A failed mux does not raise. mkvmerge's exit code and error lines come back alongside the
|
|
output path, and the caller must examine them.
|
|
|
|
Parameters:
|
|
title: Set the Matroska Container file title. Usually displayed in players
|
|
instead of the filename if set.
|
|
delete: Delete all track files after multiplexing.
|
|
progress: Update a rich progress bar with `completed=...`. This must be the
|
|
progress object's update() func, pre-set with task id by functools.partial.
|
|
audio_expected: Whether the output must have audio. unshackle uses this to
|
|
decide if it adds embedded audio metadata.
|
|
title_language: The title's intended language. Used to select the best video track
|
|
for audio metadata when multiple video tracks exist.
|
|
skip_subtitles: Skip muxing subtitle tracks into the container.
|
|
output_path: Explicit destination for the muxed container. When None (default)
|
|
unshackle derives the path from the first track, so callers muxing several track groups that
|
|
share a video list must pass distinct paths to avoid clobbering each other.
|
|
"""
|
|
if self.videos and not self.audio and audio_expected:
|
|
video_track = None
|
|
if title_language:
|
|
video_track = next((v for v in self.videos if v.language == title_language), None)
|
|
if not video_track:
|
|
video_track = next((v for v in self.videos if v.is_original_lang), None)
|
|
|
|
video_track = video_track or self.videos[0]
|
|
if video_track.language.is_valid():
|
|
lang_code = str(video_track.language)
|
|
lang_name = video_track.language.display_name()
|
|
|
|
for video in self.videos:
|
|
if video.data.get("audio_language"):
|
|
continue
|
|
video.needs_repack = True
|
|
video.data["audio_language"] = lang_code
|
|
video.data["audio_language_name"] = lang_name
|
|
|
|
if any(at.path and at.path.exists() and has_dts_uhd_sample_entry(at.path) for at in self.audio):
|
|
return self._mux_mp4(
|
|
title=title,
|
|
delete=delete,
|
|
progress=progress,
|
|
skip_subtitles=skip_subtitles,
|
|
output_path=output_path,
|
|
)
|
|
|
|
if not binaries.MKVToolNix:
|
|
raise RuntimeError("MKVToolNix (mkvmerge) is required for muxing but was not found")
|
|
|
|
cl = [
|
|
str(binaries.MKVToolNix),
|
|
"--no-date", # remove dates from the output for security
|
|
]
|
|
|
|
sample_base = self.sample_base
|
|
|
|
def sample_sync(start: float) -> list[str]:
|
|
shift = round((start - sample_base) * 1000)
|
|
return ["--sync", f"0:{shift}"] if config.sample and shift else []
|
|
|
|
if config.sample:
|
|
start, end = (seconds - sample_base for seconds in config.sample)
|
|
cl.extend(["--split", f"parts:{start:.3f}s-{end:.3f}s"])
|
|
|
|
if config.muxing.get("set_title", True):
|
|
cl.extend(["--title", title])
|
|
|
|
default_language = config.muxing.get("default_language") or {}
|
|
# mux() runs after every track is downloaded, so a typo here must not discard the title
|
|
preferred_video_lang = valid_language(default_language.get("video"))
|
|
preferred_audio_lang = valid_language(default_language.get("audio"))
|
|
preferred_subtitle_lang = valid_language(default_language.get("subtitle"))
|
|
|
|
preferred_video_idx: Optional[int] = None
|
|
if preferred_video_lang:
|
|
preferred_video_idx = next(
|
|
(idx for idx, v in enumerate(self.videos) if is_close_match(v.language, [preferred_video_lang])),
|
|
None,
|
|
)
|
|
|
|
preferred_audio_idx: Optional[int] = None
|
|
if preferred_audio_lang:
|
|
preferred_audio_idx = next(
|
|
(idx for idx, a in enumerate(self.audio) if is_close_match(a.language, [preferred_audio_lang])),
|
|
None,
|
|
)
|
|
|
|
preferred_subtitle_idx: Optional[int] = None
|
|
if preferred_subtitle_lang and not skip_subtitles:
|
|
preferred_subtitle_idx = next(
|
|
(idx for idx, s in enumerate(self.subtitles) if is_close_match(s.language, [preferred_subtitle_lang])),
|
|
None,
|
|
)
|
|
|
|
for i, vt in enumerate(self.videos):
|
|
if not vt.path or not vt.path.exists():
|
|
raise ValueError("Video Track must be downloaded before muxing...")
|
|
events.emit(events.Types.TRACK_MULTIPLEX, track=vt)
|
|
|
|
if preferred_video_idx is not None:
|
|
is_default = i == preferred_video_idx
|
|
elif title_language:
|
|
is_default = vt.language == title_language
|
|
if not any(v.language == title_language for v in self.videos):
|
|
is_default = vt.is_original_lang or i == 0
|
|
else:
|
|
is_default = i == 0
|
|
|
|
video_args = [
|
|
"--language",
|
|
f"0:{vt.language}",
|
|
"--default-track",
|
|
f"0:{is_default}",
|
|
"--original-flag",
|
|
f"0:{vt.is_original_lang}",
|
|
"--compression",
|
|
"0:none",
|
|
*sample_sync(vt.sample_start),
|
|
]
|
|
|
|
# Add FPS fix if needed (typically for hybrid mode to prevent sync issues)
|
|
if hasattr(vt, "needs_duration_fix") and vt.needs_duration_fix and vt.fps:
|
|
video_args.extend(
|
|
[
|
|
"--default-duration",
|
|
f"0:{vt.fps}fps" if isinstance(vt.fps, str) else f"0:{vt.fps:.3f}fps",
|
|
"--fix-bitstream-timing-information",
|
|
"0:1",
|
|
]
|
|
)
|
|
|
|
if hasattr(vt, "range") and vt.range == Video.Range.HLG:
|
|
video_args.extend(
|
|
[
|
|
"--color-transfer-characteristics",
|
|
"0:18", # ARIB STD-B67 (HLG)
|
|
]
|
|
)
|
|
|
|
if hasattr(vt, "data") and vt.data.get("audio_language"):
|
|
audio_lang = vt.data["audio_language"]
|
|
audio_name = vt.data.get("audio_language_name", audio_lang)
|
|
video_args.extend(
|
|
[
|
|
"--language",
|
|
f"1:{audio_lang}",
|
|
"--track-name",
|
|
f"1:{audio_name}",
|
|
]
|
|
)
|
|
|
|
cl.extend(video_args + ["(", str(vt.path), ")"])
|
|
|
|
for i, at in enumerate(self.audio):
|
|
if not at.path or not at.path.exists():
|
|
raise ValueError("Audio Track must be downloaded before muxing...")
|
|
events.emit(events.Types.TRACK_MULTIPLEX, track=at)
|
|
if preferred_audio_idx is not None:
|
|
audio_default = i == preferred_audio_idx
|
|
else:
|
|
audio_default = at.is_original_lang
|
|
cl.extend(
|
|
[
|
|
"--track-name",
|
|
f"0:{at.get_track_name() or ''}",
|
|
"--language",
|
|
f"0:{at.language}",
|
|
"--default-track",
|
|
f"0:{audio_default}",
|
|
"--visual-impaired-flag",
|
|
f"0:{at.descriptive}",
|
|
"--original-flag",
|
|
f"0:{at.is_original_lang}",
|
|
"--compression",
|
|
"0:none",
|
|
*sample_sync(at.sample_start),
|
|
"(",
|
|
str(at.path),
|
|
")",
|
|
]
|
|
)
|
|
|
|
if not skip_subtitles:
|
|
for i, st in enumerate(self.subtitles):
|
|
if not st.path or not st.path.exists():
|
|
raise ValueError("Text Track must be downloaded before muxing...")
|
|
events.emit(events.Types.TRACK_MULTIPLEX, track=st)
|
|
if preferred_subtitle_idx is not None:
|
|
default = i == preferred_subtitle_idx
|
|
else:
|
|
default = bool(self.audio and is_close_match(st.language, [self.audio[0].language]) and st.forced)
|
|
cl.extend(
|
|
[
|
|
"--track-name",
|
|
f"0:{st.get_track_name() or ''}",
|
|
"--language",
|
|
f"0:{st.language}",
|
|
"--sub-charset",
|
|
"0:UTF-8",
|
|
"--forced-track",
|
|
f"0:{st.forced}",
|
|
"--default-track",
|
|
f"0:{default}",
|
|
"--hearing-impaired-flag",
|
|
f"0:{st.sdh}",
|
|
"--original-flag",
|
|
f"0:{st.is_original_lang}",
|
|
"--compression",
|
|
"0:none",
|
|
*sample_sync(st.sample_start),
|
|
"(",
|
|
str(st.path),
|
|
")",
|
|
]
|
|
)
|
|
|
|
if self.chapters:
|
|
chapters_path = config.directories.temp / config.filenames.chapters.format(
|
|
title=sanitize_filename(title), random=f"{self.chapters.id}_{os.getpid()}_{threading.get_ident()}"
|
|
)
|
|
self.chapters.dump(chapters_path, fallback_name=config.chapter_fallback_name)
|
|
if config.sample and sample_base:
|
|
cl.extend(["--chapter-sync", str(round(-sample_base * 1000))])
|
|
cl.extend(["--chapter-charset", "UTF-8", "--chapters", str(chapters_path)])
|
|
else:
|
|
chapters_path = None
|
|
|
|
for attachment in self.attachments:
|
|
if not attachment.path or not attachment.path.exists():
|
|
raise ValueError("Attachment File was not found...")
|
|
cl.extend(
|
|
[
|
|
"--attachment-description",
|
|
attachment.description or "",
|
|
"--attachment-mime-type",
|
|
attachment.mime_type,
|
|
"--attachment-name",
|
|
attachment.name,
|
|
"--attach-file",
|
|
str(attachment.path.resolve()),
|
|
]
|
|
)
|
|
|
|
if output_path is None:
|
|
output_path = (
|
|
self.videos[0].path.with_suffix(".muxed.mkv")
|
|
if self.videos
|
|
else self.audio[0].path.with_suffix(".muxed.mka")
|
|
if self.audio
|
|
else self.subtitles[0].path.with_suffix(".muxed.mks")
|
|
if self.subtitles
|
|
else chapters_path.with_suffix(".muxed.mkv")
|
|
if self.chapters
|
|
else None
|
|
)
|
|
if not output_path:
|
|
raise ValueError("No tracks provided, at least one track must be provided.")
|
|
|
|
full_command = [*cl, "--output", str(output_path), "--gui-mode"]
|
|
|
|
log_event(
|
|
"mux_start",
|
|
level="INFO",
|
|
message=(f"Muxing {len(self.videos)}V/{len(self.audio)}A/{len(self.subtitles)}S -> {output_path.name}"),
|
|
context={
|
|
"title": title,
|
|
"output_path": str(output_path),
|
|
"muxer": str(binaries.MKVToolNix),
|
|
"command": full_command,
|
|
"video_count": len(self.videos),
|
|
"audio_count": len(self.audio),
|
|
"subtitle_count": len(self.subtitles),
|
|
"attachment_count": len(self.attachments),
|
|
"has_chapters": bool(self.chapters),
|
|
"video_tracks": [
|
|
{"id": v.id, "codec": getattr(v, "codec", None), "language": str(v.language)} for v in self.videos
|
|
],
|
|
"audio_tracks": [
|
|
{"id": a.id, "codec": getattr(a, "codec", None), "language": str(a.language)} for a in self.audio
|
|
],
|
|
"subtitle_tracks": [
|
|
{"id": s.id, "codec": getattr(s, "codec", None), "language": str(s.language)}
|
|
for s in self.subtitles
|
|
],
|
|
},
|
|
)
|
|
|
|
try:
|
|
errors = []
|
|
warnings = []
|
|
mux_start_time = time.monotonic()
|
|
p = subprocess.Popen(full_command, text=True, encoding="utf-8", errors="replace", stdout=subprocess.PIPE)
|
|
for line in iter(p.stdout.readline, ""):
|
|
if line.startswith("#GUI#error") or line.startswith("#GUI#warning"):
|
|
errors.append(line)
|
|
if line.startswith("#GUI#warning"):
|
|
warnings.append(line.strip())
|
|
if "progress" in line:
|
|
progress(total=100, completed=int(line.strip()[14:-1]))
|
|
|
|
returncode = p.wait()
|
|
mux_duration_ms = round((time.monotonic() - mux_start_time) * 1000, 1)
|
|
output_size = output_path.stat().st_size if output_path and output_path.exists() else 0
|
|
|
|
if returncode != 0 or errors:
|
|
log_event(
|
|
"mux_failed",
|
|
level="ERROR",
|
|
message=f"mkvmerge exited with code {returncode}",
|
|
context={
|
|
"returncode": returncode,
|
|
"output_path": str(output_path),
|
|
"errors": errors,
|
|
"warnings": warnings,
|
|
"duration_ms": mux_duration_ms,
|
|
},
|
|
)
|
|
else:
|
|
log_event(
|
|
"mux_complete",
|
|
level="INFO",
|
|
message=(
|
|
f"Muxed {output_path.name} ({output_size} bytes) in {mux_duration_ms}ms"
|
|
+ (f" with {len(warnings)} warning(s)" if warnings else "")
|
|
),
|
|
context={
|
|
"output_path": str(output_path),
|
|
"output_exists": output_path.exists() if output_path else False,
|
|
"output_size": output_size,
|
|
"duration_ms": mux_duration_ms,
|
|
"returncode": returncode,
|
|
"warnings": warnings,
|
|
},
|
|
)
|
|
|
|
return output_path, returncode, errors
|
|
finally:
|
|
if chapters_path:
|
|
chapters_path.unlink()
|
|
if delete:
|
|
for track in self:
|
|
track.delete()
|
|
for attachment in self.attachments:
|
|
attachment.delete()
|
|
|
|
|
|
__all__ = ("Tracks",)
|