feat(dl): add --audio-description-only and --forced-subs-only

Each keeps one kind of track inside its track type. A run of more than one title skips a title that has no such track. The REST API rejects a non-boolean value for a boolean dl param, and the DASH parser reads audio description from Role elements.

tests/core/test_dash_descriptive_audio.py

@@ -37,3 +37,13 @@ def test_non_descriptive_accessibility_values(scheme: str, value: str) -> None:
 
 def test_no_accessibility_element() -> None:
     assert DASH.is_descriptive(ET.fromstring("<AdaptationSet/>")) is False
+
+
+def test_descriptive_role_element() -> None:
+    # some manifests mark audio description with a Role and no Accessibility element
+    element = ET.fromstring(
+        '<AdaptationSet><Role schemeIdUri="urn:mpeg:dash:role:2011" value="description"/></AdaptationSet>'
+    )
+    assert DASH.is_descriptive(element) is True
+    main = ET.fromstring('<AdaptationSet><Role schemeIdUri="urn:mpeg:dash:role:2011" value="main"/></AdaptationSet>')
+    assert DASH.is_descriptive(main) is False

tests/lang_selection/test_kind_only_flags.py

@@ -0,0 +1,90 @@
+"""The kind-only flags: --audio-description-only and --forced-subs-only.
+
+The filters in dl.result are not extractable without executing the whole download loop, so
+their wiring is pinned against the source, the same way test_required_langs_gate does it.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from unshackle.commands.dl import dl
+from unshackle.core.api.handlers import validate_download_parameters
+from unshackle.core.utilities import kind_only_conflict
+
+KEYS = ["audio_description_only", "forced_subs_only"]
+BASE = {"service": "EXAMPLE", "title": "x"}
+
+
+@pytest.mark.parametrize(
+    "args,only_key,base_key,expected",
+    [
+        (["-ado"], "audio_description_only", "audio_description", (True, False)),
+        (["-ad"], "audio_description_only", "audio_description", (False, True)),
+        ([], "audio_description_only", "audio_description", (False, False)),
+        (["-fso"], "forced_subs_only", "forced_subs", (True, False)),
+        (["-fs"], "forced_subs_only", "forced_subs", (False, True)),
+    ],
+)
+def test_only_flag_does_not_parse_as_its_base_flag(args, only_key, base_key, expected):
+    params = dl.cli.make_context("dl", args, resilient_parsing=True).params
+    assert (params[only_key], params[base_key]) == expected
+
+
+@pytest.mark.parametrize(
+    "params,expected",
+    [
+        ({"audio_description_only": True, "audio_only": True}, None),
+        ({"audio_description_only": True}, None),
+        ({"audio_description_only": True, "no_audio": True}, ("audio_description_only", "no_audio")),
+        ({"audio_description_only": True, "video_only": True}, ("audio_description_only", "video_only")),
+        ({"forced_subs_only": True, "subs_only": True}, None),
+        ({"forced_subs_only": True, "no_subs": True}, ("forced_subs_only", "no_subs")),
+        ({"forced_subs_only": True, "audio_only": True}, ("forced_subs_only", "audio_only")),
+        ({"no_audio": True, "video_only": True}, None),
+        # the *_only track-type flags combine: -A -S keeps audio, so -ado is not dropped
+        ({"audio_description_only": True, "audio_only": True, "subs_only": True}, None),
+        ({"forced_subs_only": True, "subs_only": True, "audio_only": True}, None),
+        ({"audio_description_only": True, "subs_only": True}, ("audio_description_only", "subs_only")),
+    ],
+)
+def test_kind_only_conflict(params, expected):
+    assert kind_only_conflict(params) == expected
+
+
+def test_api_rejects_a_conflict():
+    assert validate_download_parameters({**BASE, "audio_description_only": True, "no_audio": True}) == (
+        "Cannot use both audio_description_only and no_audio"
+    )
+    assert validate_download_parameters({**BASE, "audio_description_only": True, "audio_only": True}) is None
+
+
+@pytest.mark.parametrize("key", [*KEYS, "audio_description", "audio_only"])
+def test_api_rejects_a_non_boolean_flag(key):
+    # the string "false" is truthy in dl.result, so it would turn the flag on
+    assert validate_download_parameters({**BASE, key: "false"}) == f"{key} must be a boolean"
+    assert validate_download_parameters({**BASE, key: None}) is None
+
+
+class TestFilterWiring:
+    source = inspect.getsource(dl.result)
+
+    def test_ado_runs_when_the_title_has_no_separate_audio(self):
+        # embedded audio is standard audio: the filter must sit before the `len(...) > 0` guard
+        ado = self.source.index("if keep_audio and audio_description_only:")
+        guard = self.source.index("if keep_audio and len(title.tracks.audio) > 0:")
+        assert ado < guard
+
+    def test_ado_does_not_count_embedded_audio(self):
+        # both the --require-audio gate and the missing-language check
+        assert self.source.count("keep_videos and not audio_description_only") == 2
+
+    def test_only_filters_keep_one_kind(self):
+        assert "title.tracks.select_audio(lambda x: x.descriptive)" in self.source
+        assert "title.tracks.select_subtitles(lambda x: x.forced)" in self.source
+
+    def test_fso_implies_forced_subs(self):
+        # services read forced_subs from the ctx params to fetch forced tracks at all
+        assert "if forced_s_lang or fsl_excl or forced_subs_only:" in self.source

tests/lang_selection/test_kind_only_selection.py

@@ -0,0 +1,149 @@
+"""Run the kind-only filters through the real dl.result() selection loop.
+
+A fake service hands in-memory tracks to the loop, and the run stops at the first
+title that reaches ``Tracks.tree``, the step right after selection. A title with no
+track of the kind never gets there, so what the hook records is what the run kept.
+"""
+
+from __future__ import annotations
+
+import inspect
+import logging
+from typing import Any, cast
+
+import pytest
+
+from unshackle.commands.dl import dl
+from unshackle.core.api.handlers import DEFAULT_DOWNLOAD_PARAMS
+from unshackle.core.service import Service
+from unshackle.core.titles import Episode, Series
+from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
+
+
+class Selected(BaseException):
+    """Carries the kept track ids out of the loop."""
+
+    def __init__(self, tracks: Tracks) -> None:
+        self.ids = {
+            "video": [t.id for t in tracks.videos],
+            "audio": [t.id for t in tracks.audio],
+            "subs": [t.id for t in tracks.subtitles],
+        }
+
+
+class Harness(dl):
+    def __getattr__(self, name: str) -> None:
+        # every optional dl attribute (tmdb ids, cdm, vaults, ...) reads as unset
+        return None
+
+
+class FakeService:
+    """One episode per entry; ``ad`` adds a descriptive track, ``forced`` a forced subtitle."""
+
+    def __init__(self, episodes: dict[int, dict[str, bool]]) -> None:
+        self.episodes = episodes
+        self.session = None
+
+    def authenticate(self, cookies: Any = None, credential: Any = None) -> None:
+        pass
+
+    def get_titles_cached(self) -> Series:
+        return Series(
+            Episode(id_=f"show-s01-e{n:03}", service=FakeService, title="Show", season=1, number=n, language="en")
+            for n in self.episodes
+        )
+
+    def get_tracks(self, title: Episode) -> Tracks:
+        kinds = self.episodes[title.number]
+        tracks = Tracks()
+        tracks.add(Video(id_="v", url="https://x/v", language="en", codec=Video.Codec.AVC, range_=Video.Range.SDR))
+        tracks.add(Audio(id_="a", url="https://x/a", language="en", codec=Audio.Codec.AAC, bitrate=128_000))
+        if kinds.get("ad"):
+            tracks.add(
+                Audio(id_="ad", url="https://x/ad", language="en", codec=Audio.Codec.AAC, bitrate=1, descriptive=True)
+            )
+        tracks.add(Subtitle(id_="s", url="https://x/s", language="en", codec=Subtitle.Codec.WebVTT))
+        if kinds.get("forced"):
+            tracks.add(Subtitle(id_="sf", url="https://x/sf", language="en", codec=Subtitle.Codec.WebVTT, forced=True))
+        return tracks
+
+    def get_chapters(self, title: Episode) -> list:
+        return []
+
+    def __getattr__(self, name: str) -> Any:
+        if name.startswith("on_"):
+            # the event hooks (on_track_downloaded, ...) the loop subscribes
+            return lambda *a, **k: None
+        raise AttributeError(name)
+
+
+def run(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, episodes: dict, **flags: Any) -> Any:
+    """Return the Selected ids, the exit code, or None for a run that completes."""
+    monkeypatch.setattr(Tracks, "tree", lambda self, *a, **k: (_ for _ in ()).throw(Selected(self)))
+    cmd = Harness.__new__(Harness)
+    cmd.log = logging.getLogger("test_kind_only")
+    cmd.service = "FAKE"
+    cmd.tmdb_searched = True
+    cmd.completed_files = []
+    kwargs: dict[str, Any] = {}
+    for name, param in inspect.signature(dl.result).parameters.items():
+        if name in DEFAULT_DOWNLOAD_PARAMS:
+            kwargs[name] = DEFAULT_DOWNLOAD_PARAMS[name]
+        elif name not in ("self", "service"):
+            kwargs[name] = None if param.default is inspect.Parameter.empty else param.default
+    kwargs.update(quality=[], vcodec=[], acodec=[], range_=[], wanted=[], cdm_only=None, downloads=1)
+    kwargs.update(flags)
+    caplog.set_level(logging.WARNING, logger="test_kind_only")
+    try:
+        cmd.result(cast(Service, FakeService(episodes)), **kwargs)
+    except Selected as sel:
+        return sel.ids
+    except SystemExit as exc:
+        return exc.code
+    return None
+
+
+def test_ado_keeps_the_descriptive_track_and_the_other_types(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, audio_description_only=True)
+    assert ids == {"video": ["v"], "audio": ["ad"], "subs": ["s"]}
+
+
+def test_fso_keeps_the_forced_track_and_the_other_types(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, forced_subs_only=True)
+    assert ids == {"video": ["v"], "audio": ["a"], "subs": ["sf"]}
+
+
+def test_only_flags_with_their_type_flags_give_a_single_kind(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, audio_description_only=True, audio_only=True)
+    assert ids == {"video": [], "audio": ["ad"], "subs": []}
+    ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, forced_subs_only=True, subs_only=True)
+    assert ids == {"video": [], "audio": [], "subs": ["sf"]}
+
+
+@pytest.mark.parametrize("flag", ["audio_description_only", "forced_subs_only"])
+def test_a_single_title_with_no_such_track_exits(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, flag: str
+) -> None:
+    assert run(monkeypatch, caplog, {1: {}}, **{flag: True}) == 1
+    assert "Skipping" not in caplog.text
+
+
+def test_a_multi_title_run_skips_the_title_and_continues(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    ids = run(monkeypatch, caplog, {1: {}, 2: {"ad": True}}, audio_description_only=True)
+    assert ids["audio"] == ["ad"]
+    assert "Skipping Show S01E01" in caplog.text
+
+
+def test_a_multi_title_run_with_every_title_skipped_completes(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    assert run(monkeypatch, caplog, {1: {}, 2: {}}, forced_subs_only=True) is None
+    assert caplog.text.count("Skipping Show S01E") == 2

tests/lang_selection/test_required_langs.py

@@ -5,6 +5,7 @@ import inspect
 import pytest
 
 from unshackle.commands.dl import dl
+from unshackle.core.api import download_manager
 from unshackle.core.api.handlers import DEFAULT_DOWNLOAD_PARAMS
 from unshackle.core.utilities import missing_required_langs
 
@@ -102,15 +103,15 @@ def test_api_accepts_the_require_flags():
     )
 
 
-@pytest.mark.parametrize("key", ["require_audio", "require_video"])
-def test_api_reaches_dl_result(key, monkeypatch):
+@pytest.mark.parametrize(
+    "key,default",
+    [("require_audio", []), ("require_video", []), ("audio_description_only", False), ("forced_subs_only", False)],
+)
+def test_api_reaches_dl_result(key, default):
     # download_manager builds the dl.result kwargs by hand, so a missing key is a silent drop
-    import inspect as _inspect
-
-    from unshackle.core.api import download_manager
-
-    source = _inspect.getsource(download_manager)
-    assert f'{key}=params.get("{key}", [])' in source
+    assert DEFAULT_DOWNLOAD_PARAMS[key] == default
+    assert key in inspect.signature(dl.result).parameters
+    assert f'{key}=params.get("{key}", {default!r})' in inspect.getsource(download_manager)
 
 
 @pytest.mark.parametrize(

tests/lang_selection/test_required_langs_gate.py

@@ -48,7 +48,7 @@ class TestGateWiring:
         assert "require_video if keep_videos else []" in self.source
 
     def test_gate_counts_audio_embedded_in_kept_video(self):
-        assert "embedded_audio_langs(title.tracks.videos, keep_videos)" in self.source
+        assert "embedded_audio_langs(title.tracks.videos, keep_videos and not audio_description_only)" in self.source
 
     def test_gate_honours_exact_lang(self):
         assert "missing_required_langs(required, available, title.language, exact=exact_lang)" in self.source