diff --git a/tests/conftest.py b/tests/conftest.py index 5d03a24b..cf15bab4 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,10 +4,12 @@ from __future__ import annotations import socket import socketserver -from typing import Any +from pathlib import Path +from typing import Any, Callable, Optional import pytest +from unshackle.core import binaries from unshackle.core.config import config from unshackle.core.update_checker import UpdateChecker @@ -45,3 +47,27 @@ def no_update_check(monkeypatch: pytest.MonkeyPatch) -> None: def default_muxing(monkeypatch: pytest.MonkeyPatch) -> None: """Mux with the default options, whatever the muxing block of the local unshackle.yaml sets.""" monkeypatch.setattr(config, "muxing", {}) + + +@pytest.fixture +def mux_args(monkeypatch: pytest.MonkeyPatch) -> Callable[..., list[str]]: + """Run Tracks.mux up to mkvmerge and return its arguments; ``ids`` is what the track ID probe reports.""" + + class Stop(Exception): + pass + + def run(tracks: Any, ids: Optional[dict[str, int]] = None, **kwargs: Any) -> list[str]: + captured: list[list[str]] = [] + + def fake_popen(args: list[str], **_: object) -> None: + captured.append(args) + raise Stop + + monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen) + monkeypatch.setattr("unshackle.core.tracks.tracks.file_track_ids", lambda path: ids or {"video": 0}) + monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge")) + with pytest.raises(Stop): + tracks.mux("Title", delete=False, **kwargs) + return captured[0] + + return run diff --git a/tests/test_sample.py b/tests/test_sample.py index 160c3a47..368048b2 100644 --- a/tests/test_sample.py +++ b/tests/test_sample.py @@ -8,7 +8,7 @@ import struct import threading from pathlib import Path from types import SimpleNamespace -from typing import Optional +from typing import Callable, Optional from unittest.mock import MagicMock import click @@ -16,7 +16,6 @@ import pytest from requests import Session from unshackle.commands.dl import sample_name -from unshackle.core import binaries from unshackle.core.api.handlers import validate_download_parameters from unshackle.core.config import config from unshackle.core.manifests.dash import DASH @@ -231,19 +230,7 @@ def test_subtitle_trim_leaves_an_unreadable_file_whole(tmp_path: Path) -> None: assert subtitle.path.read_bytes() == b"\x00\x01not a subtitle" -def mux_command(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[str]: - class Stop(Exception): - pass - - captured: list[list[str]] = [] - - def fake_popen(args: list[str], **kwargs: object) -> None: - captured.append(args) - raise Stop - - monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen) - monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge")) - +def mux_command(tmp_path: Path, mux_args: Callable[..., list[str]]) -> list[str]: video, audio = make_video(), make_audio() trimmed = Subtitle(id_="s1", url="https://x/s1", codec=Subtitle.Codec.SubRip, language="en") whole = Subtitle(id_="s2", url="https://x/s2", codec=Subtitle.Codec.SubRip, language="fr") @@ -254,14 +241,14 @@ def mux_command(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[str]: tracks = Tracks([video, audio, trimmed, whole]) tracks.chapters.add(Chapter("00:02:05.000", "Scene")) - with pytest.raises(Stop): - tracks.mux("Title", delete=False, progress=lambda **k: None, output_path=tmp_path / "out.mkv") - return captured[0] + return mux_args(tracks, progress=lambda **k: None, output_path=tmp_path / "out.mkv") -def test_the_mux_shifts_each_track_and_cuts_to_the_window(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_the_mux_shifts_each_track_and_cuts_to_the_window( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mux_args: Callable[..., list[str]] +) -> None: monkeypatch.setattr(config, "sample", (121, 181)) - command = mux_command(tmp_path, monkeypatch) + command = mux_command(tmp_path, mux_args) def sync_before(name: str) -> str: file_at = command.index(str(tmp_path / name)) @@ -273,17 +260,19 @@ def test_the_mux_shifts_each_track_and_cuts_to_the_window(tmp_path: Path, monkey return command[sync_at + 1] assert command[command.index("--split") + 1] == "parts:1.500s-61.500s" - assert sync_before("v.mp4") == "0:500" + assert sync_before("v.mp4") == "-1:500" assert "--sync" not in command[: command.index(str(tmp_path / "a.mp4"))][-12:], "the earliest track is not moved" - assert sync_before("s1.srt") == "0:1500" - assert sync_before("s2.srt") == "0:-119500" + assert sync_before("s1.srt") == "-1:1500" + assert sync_before("s2.srt") == "-1:-119500" assert command[command.index("--chapter-sync") + 1] == "-119500" assert command.index("--chapter-sync") < command.index("--chapters") -def test_a_full_download_adds_no_sample_options(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: +def test_a_full_download_adds_no_sample_options( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mux_args: Callable[..., list[str]] +) -> None: monkeypatch.setattr(config, "sample", None) - command = mux_command(tmp_path, monkeypatch) + command = mux_command(tmp_path, mux_args) assert not {"--split", "--sync", "--chapter-sync"} & set(command) diff --git a/tests/tracks/test_mux_embedded_audio_metadata.py b/tests/tracks/test_mux_embedded_audio_metadata.py index 83191c97..9e602ec4 100644 --- a/tests/tracks/test_mux_embedded_audio_metadata.py +++ b/tests/tracks/test_mux_embedded_audio_metadata.py @@ -1,7 +1,11 @@ +import json +from types import SimpleNamespace + import pytest from unshackle.core import binaries from unshackle.core.tracks import Tracks, Video +from unshackle.core.tracks.tracks import file_track_ids def make_video(track_id: str, **data) -> Video: @@ -19,6 +23,12 @@ def make_video(track_id: str, **data) -> Video: return video +def downloaded(video: Video, tmp_path) -> Video: + video.path = tmp_path / "v.ts" + video.path.write_bytes(b"\x00" * 16) + return video + + def mux_until_mkvmerge_check(tracks: Tracks) -> None: # the embedded-audio metadata pass runs before the mkvmerge lookup, so a missing # binary stops the mux right after the state under test has been written @@ -35,7 +45,7 @@ def test_names_undeclared_video(monkeypatch): assert video.needs_repack assert video.data["audio_language"] == "zh" - assert video.data["audio_language_name"] == "Chinese" + assert video.embedded_audio_name() == "Chinese" def test_service_declaration_is_preserved(monkeypatch): @@ -50,3 +60,36 @@ def test_service_declaration_is_preserved(monkeypatch): assert declared.data["audio_language_name"] == "Thai [Original]" assert not declared.needs_repack assert undeclared.data["audio_language"] == "zh" + + +def test_embedded_audio_without_a_name_gets_the_language_name(tmp_path, mux_args): + args = mux_args(Tracks(downloaded(make_video("v1", audio_language="pt-BR"), tmp_path)), {"video": 0, "audio": 1}) + assert args[args.index("--track-name") + 1] == "1:Portuguese" + + +def test_embedded_audio_options_use_the_file_track_ids(tmp_path, mux_args): + video = make_video("v1", audio_language="ko", audio_language_name="Korean") + args = mux_args(Tracks(downloaded(video, tmp_path)), {"audio": 0, "video": 1}) + assert args[args.index("--track-name") + 1] == "0:Korean" + original = f"{video.is_original_lang}" + assert [args[i + 1] for i, arg in enumerate(args) if arg == "--original-flag"] == [f"1:{original}", f"0:{original}"] + assert "0:ko" in args and "1:zh" in args + + +def test_embedded_audio_options_are_left_out_when_the_file_has_no_audio(tmp_path, mux_args): + args = mux_args(Tracks(downloaded(make_video("v1", audio_language="ko"), tmp_path)), {"video": 0}) + assert "--track-name" not in args + + +def test_repack_and_mux_share_the_embedded_audio_name(): + assert make_video("v1", audio_language="zh").embedded_audio_name() == "Chinese" + assert make_video("v1", audio_language="zh", audio_language_name="Mandarin").embedded_audio_name() == "Mandarin" + + +def test_file_track_ids_reads_the_first_id_of_each_type(monkeypatch, tmp_path): + tracks = [{"id": 0, "type": "audio"}, {"id": 1, "type": "video"}, {"id": 2, "type": "audio"}] + reply = SimpleNamespace(stdout=json.dumps({"tracks": tracks})) + monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.run", lambda *a, **k: reply) + assert file_track_ids(tmp_path / "v.ts") == {"audio": 0, "video": 1} + monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.run", lambda *a, **k: SimpleNamespace(stdout="")) + assert file_track_ids(tmp_path / "v.ts") == {"video": 0, "audio": 1} diff --git a/tests/tracks/test_track_language_names.py b/tests/tracks/test_track_language_names.py index 80c91300..beb4edda 100644 --- a/tests/tracks/test_track_language_names.py +++ b/tests/tracks/test_track_language_names.py @@ -1,10 +1,9 @@ from pathlib import Path -from typing import Optional +from typing import Callable, Optional import pytest from langcodes import Language -from unshackle.core import binaries from unshackle.core.config import config from unshackle.core.tracks import Audio, Subtitle, Tracks, Video from unshackle.core.utilities import language_name, multi_script_languages, with_script @@ -130,18 +129,7 @@ def test_a_service_name_wins() -> None: assert sub("zh", name="简体中文").get_track_name() == "简体中文" -def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - class Stop(Exception): - pass - - captured: list[list[str]] = [] - - def fake_popen(args: list[str], **kwargs: object) -> None: - captured.append(args) - raise Stop - - monkeypatch.setattr("unshackle.core.tracks.tracks.subprocess.Popen", fake_popen) - monkeypatch.setattr(binaries, "MKVToolNix", Path("mkvmerge")) +def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, mux_args: Callable[..., list[str]]) -> None: video = Video(id_="v", url="https://x/v", codec=Video.Codec.AVC, language="zh-TW", width=1920, height=1080) audio = Audio(id_="a", url="https://x/a", codec=Audio.Codec.AAC, language="zh-TW", bitrate=1) subtitle = sub("zh-TW") @@ -149,12 +137,11 @@ def test_mkvmerge_gets_the_subtitle_script_and_name(tmp_path: Path, monkeypatch: track.path = tmp_path / file track.path.write_bytes(b"\x00" * 16) - with pytest.raises(Stop): - Tracks([video, audio, subtitle]).mux("T", delete=False, output_path=tmp_path / "o.mkv") + args = mux_args(Tracks([video, audio, subtitle]), output_path=tmp_path / "o.mkv") def last_option(file: str, flag: str) -> str: """Return the value of the last flag given before the file, which is the one mkvmerge applies to it.""" - before = captured[0][: captured[0].index(str(tmp_path / file))] + before = args[: args.index(str(tmp_path / file))] return before[len(before) - before[::-1].index(flag)] assert last_option("s.vtt", "--language") == "0:zh-Hant-TW" diff --git a/unshackle/core/tracks/track.py b/unshackle/core/tracks/track.py index f174d09c..6e14cd32 100644 --- a/unshackle/core/tracks/track.py +++ b/unshackle/core/tracks/track.py @@ -1012,6 +1012,11 @@ class Track: """The language tag written to the output file.""" return self.language + def embedded_audio_name(self) -> str: + """Name the audio a service keeps inside this track's file, by its language when the service set none.""" + lang = self.data["audio_language"] + return self.data.get("audio_language_name") or language_name(Language.get(lang)) or lang + def get_track_name(self) -> Optional[str]: """Get the Track Name, or the name of its language when the service set none.""" return self.name if self.name is not None else language_name(self.mux_language) @@ -1276,7 +1281,7 @@ class Track: if hasattr(self, "data") and self.data.get("audio_language"): audio_lang = self.data["audio_language"] - audio_name = self.data.get("audio_language_name", audio_lang) + audio_name = self.embedded_audio_name() args.extend( [ "-metadata:s:a:0", diff --git a/unshackle/core/tracks/tracks.py b/unshackle/core/tracks/tracks.py index ea397d92..23d1d05e 100644 --- a/unshackle/core/tracks/tracks.py +++ b/unshackle/core/tracks/tracks.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import logging import os import re @@ -28,7 +29,6 @@ from unshackle.core.tracks.track import Track, has_dts_uhd_sample_entry, strip_d from unshackle.core.tracks.video import Video from unshackle.core.utilities import ( is_close_match, - language_name, log_event, matching_languages, original_tag, @@ -47,6 +47,21 @@ def at_resolution(video: Video, res: int) -> bool: return video.height == res or bool(video.width and int(video.width * 9 / 16) == res) +def file_track_ids(path: Path) -> dict[str, int]: + """Return the mkvmerge ID of the first track of each type in a file. A TS file can hold its audio first.""" + res = subprocess.run( + [str(binaries.MKVToolNix), "-J", str(path)], capture_output=True, text=True, encoding="utf-8", errors="replace" + ) + ids: dict[str, int] = {} + for track in json.loads(res.stdout or "{}").get("tracks") or []: + ids.setdefault(track["type"], track["id"]) + if not ids: + logging.getLogger("Tracks").debug( + f"mkvmerge found no tracks in {path.name}, so it is addressed as video 0, audio 1" + ) + return ids or {"video": 0, "audio": 1} + + class Tracks: """ Video, Audio, Subtitle, Chapter, and Attachment Track Store. @@ -766,14 +781,11 @@ class Tracks: video_track = video_track or self.videos[0] if video_track.language.is_valid(): lang_code = str(video_track.language) - lang_name = language_name(video_track.language) or lang_code - 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( @@ -796,7 +808,7 @@ class Tracks: def sample_sync(start: float) -> list[str]: shift = round((start - sample_base) * 1000) - return ["--sync", f"0:{shift}"] if config.sample and shift else [] + return ["--sync", f"-1:{shift}"] if config.sample and shift else [] if config.sample: start, end = (seconds - sample_base for seconds in config.sample) @@ -831,15 +843,17 @@ class Tracks: else: is_default = i == 0 + ids = file_track_ids(vt.path) + v = ids.get("video", 0) video_args = [ "--language", - f"0:{vt.mux_language}", + f"{v}:{vt.mux_language}", "--default-track", - f"0:{is_default}", + f"{v}:{is_default}", "--original-flag", - f"0:{vt.is_original_lang}", + f"{v}:{vt.is_original_lang}", "--compression", - "0:none", + f"{v}:none", *sample_sync(vt.sample_start), ] @@ -848,9 +862,9 @@ class Tracks: video_args.extend( [ "--default-duration", - f"0:{vt.fps}fps" if isinstance(vt.fps, str) else f"0:{vt.fps:.3f}fps", + f"{v}:{vt.fps}fps" if isinstance(vt.fps, str) else f"{v}:{vt.fps:.3f}fps", "--fix-bitstream-timing-information", - "0:1", + f"{v}:1", ] ) @@ -858,19 +872,20 @@ class Tracks: video_args.extend( [ "--color-transfer-characteristics", - "0:18", # ARIB STD-B67 (HLG) + f"{v}: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) + a = ids.get("audio") + if vt.data.get("audio_language") and a is not None: video_args.extend( [ "--language", - f"1:{audio_lang}", + f"{a}:{vt.data['audio_language']}", "--track-name", - f"1:{audio_name}", + f"{a}:{vt.embedded_audio_name()}", + "--original-flag", + f"{a}:{vt.is_original_lang}", ] )